diff --git a/README.md b/README.md index 1221348..952c543 100644 --- a/README.md +++ b/README.md @@ -13,8 +13,8 @@ A股数据抓取工具，以 [BaoStock](http://baostock.com) 为主、新浪/腾 | 涨跌停统计 | 每日主板（10%）/ 科创创业板（20%）涨跌停数量 | stock_daily 汇总 | | 季频财务指标 | 盈利能力、偿债能力、现金流（最近8个季度，JSON 存储） | BaoStock | | 分红送转 | 每10股送转、派息、除权除息日（最近10年） | BaoStock | -| 分钟K线 | 5/15/30/60 分钟K线（开高低收、成交量/额） | BaoStock | -| 概念板块 | 东方财富全量概念板块及其成分股（已有≥300个概念时自动跳过） | 东方财富 | +| 分钟K线 | 5分钟K线（开高低收、成交量/额） | 通达信本地客户端（默认） / BaoStock（可选） | +| 概念板块 | 通达信本地概念板块缓存，缺失时回退同花顺 | 通达信本地 / 同花顺 | **已知限制**：BaoStock 不含北交所（920xxx）股票；新浪/腾讯/东方财富数据源对北交所同样不支持。 @@ -31,7 +31,10 @@ A股数据抓取工具，以 [BaoStock](http://baostock.com) 为主、新浪/腾 pip install baostock akshare pymysql sqlalchemy pyyaml pandas requests ``` -> 说明：`baostock` 为主要数据源；`akshare` 用于新浪日线兜底；`requests` 用于腾讯/东方财富/概念板块抓取。 +> 说明：`baostock` 为主要数据源；`akshare` 用于新浪日线兜底和同花顺概念板块列表；`requests` 用于腾讯/同花顺概念板块抓取；通达信概念板块来自本地缓存文件。 +> 通达信分钟缓存刷新已自动按最多 100 只股票一批拆分；`--tdx-cache` 会先预热本地缓存，再全历史回填分钟表。`--tdx-local-cache` 会在预热后自动校验是否真的可读；若只想单独确认缓存是否能被读到，请用 `--tdx-verify-cache`。注意通达信客户端原生只支持刷新 `1m/5m` 本地缓存。 + +> 通达信本地目录默认读取 `config.yaml` 里的 `tdx.dir`。 ### 3. 配置数据库 @@ -58,6 +61,9 @@ fetch: delay: 0.1 # 请求间隔(秒) retry: 2 # 失败重试次数 workers: 1 # 并发线程数 + +tdx: + dir: "C:\\new_tdx64" # 通达信安装目录 ``` 确保 MySQL 中已创建对应数据库： @@ -103,13 +109,25 @@ python -m src.main --dividend # 抓取单只股票的分红 python -m src.main --dividend --symbol 000001 -# 抓取分钟K线（默认5分钟，近30天） +# 抓取分钟K线（默认5分钟，近30天，默认走通达信本地客户端） +python -m src.main --intraday + +# 使用通达信本地客户端抓取分钟K线（默认） +python -m src.main --intraday --intraday-source tdx + +# 全历史回填通达信分钟数据（先预热本地缓存，再写入分钟表） +python -m src.main --tdx-cache + +# 仅预热通达信本地分钟缓存，并自动校验，不写数据库（原生只刷 1m/5m） +python -m src.main --tdx-local-cache + +# 只验证通达信本地分钟缓存是否可读 +python -m src.main --tdx-verify-cache + +# 抓取5分钟K线 python -m src.main --intraday --start-date 20260101 -# 抓取全部频率分钟K线（5/15/30/60） -python -m src.main --intraday --freq all --start-date 20260101 - -# 抓取概念板块及成分股（东方财富） +# 抓取概念板块及成分股（通达信本地，缺失时回退同花顺） python -m src.main --sector ``` @@ -127,8 +145,12 @@ python -m src.main --sector --financial 抓取季频财务指标（增量；只抓缺失季度） --dividend 抓取分红送转数据（增量；只抓缺失年份） --intraday 抓取分钟K线行情（增量；只抓缺失日期） - --freq K线频率: 5/15/30/60/all（默认 5） - --sector 抓取概念板块及成分股（增量；已有≥300个概念时跳过） + --intraday-source 分钟K线数据源：tdx/baostock（默认 tdx） + --tdx-cache 全历史回填通达信5分钟数据（先预热本地缓存，再写入分钟表） + --tdx-local-cache 仅预热通达信本地分钟缓存，并自动校验，不写数据库（原生只刷 1m/5m） + --tdx-verify-cache 只验证通达信本地分钟缓存是否可读，不写数据库 + --freq K线频率: 5（默认 5） + --sector 抓取概念板块及成分股（增量；本地板块已齐全时跳过） --market-daily 汇总每日涨跌停统计（从 stock_daily 聚合） 日期过滤（对日线行情、交易日历、分钟K线、指数生效）: @@ -231,7 +253,7 @@ python -m src.main --sector 联合主键：`(code, date)` -### stock_min5 / stock_min15 / stock_min30 / stock_min60 — 分钟K线 +### stock_min5 — 分钟K线 | 字段 | 类型 | 说明 | |------|------|------| @@ -251,7 +273,7 @@ python -m src.main --sector | 字段 | 类型 | 说明 | |------|------|------| | code | VARCHAR(10) | 股票代码 | -| concept_code | VARCHAR(20) | 概念板块代码（东方财富 BK 编码） | +| concept_code | VARCHAR(20) | 概念板块代码（同花顺 BK 编码） | | concept_name | VARCHAR(100) | 概念板块名称 | 联合主键：`(code, concept_code)` @@ -311,8 +333,8 @@ ashare-data/ │ ├── financial.py # 季频财务指标（盈利/偿债/现金流，JSON 存储） │ ├── dividend.py # 分红送转 │ ├── intraday.py # 分钟K线（5/15/30/60） -│ └── sector.py # 概念板块及成分股（东方财富） -├── tests/ # pytest 测试（42 个用例，纯函数 + mock） +│ └── sector.py # 概念板块及成分股（通达信本地 + 同花顺） +├── tests/ # pytest 测试（48 个用例，纯函数 + mock） ├── benchmarks/ # 并发度压测脚本（不在 CI 跑，需真实 MySQL+外网） └── gzl/ # 选股脚本（独立子项目，可选） ├── Selector.py @@ -321,10 +343,10 @@ ashare-data/ ## 设计说明 -- **多数据源容灾**：日线行情默认 `--source all` 在 BaoStock / 新浪 / 腾讯 / 东方财富之间轮换并自动切换，单源失败不影响整体进度；其他模块（财务、分红、指数、分钟K线）仍以 BaoStock 为主。 +- **多数据源容灾**：日线行情默认 `--source all` 在 BaoStock / 新浪 / 腾讯 / 东方财富之间轮换并自动切换，单源失败不影响整体进度；其他模块（财务、分红、指数）仍以 BaoStock 为主，分钟K线可选本地通达信客户端。 - **线程安全**：BaoStock 的 `query_xxx()` 非线程安全，所有调用通过 `src/baostock_conn.py` 的全局锁串行化；查询超时/连接断开时自动重连。 - **去重写入**：所有表通过 `db.batch_upsert()` 走 MySQL `INSERT ON DUPLICATE KEY UPDATE`，重复执行不会产生重复数据。 -- **全量增量**：所有模块均支持增量更新——日线/指数/分钟K线按交易日对比找缺口，财务按缺失季度，分红按缺失年份，概念板块已有≥300个时跳过。 +- **全量增量**：所有模块均支持增量更新——日线/指数/分钟K线按交易日对比找缺口，财务按缺失季度，分红按缺失年份，概念板块在本地已齐全时跳过。 - **停牌识别**：日线行情若两端已覆盖、内部仍有缺口，则视为停牌，不再重抓。 - **表结构自动迁移**：`init_db()` 会检查并升级旧版 `stock_no_data` / `stock_sector` / `stock_intraday` 的列定义，无需手动改库。 - **进程内缓存**：`get_stock_codes()` / `get_ipo_dates()` 缓存全量股票代码与上市日期，避免重复扫库。 @@ -337,12 +359,14 @@ pip install pytest # 或 pip install -e ".[dev]" pytest -v ``` -当前 42 个用例全部通过，覆盖： +当前 59 个用例全部通过，覆盖： - `baostock_conn.code_to_bs` / `daily._code_to_*` 各数据源代码前缀映射 - `daily._fill_derived_fields` 振幅/涨跌幅/涨跌额补算 - `_fetch_tencent` / `_fetch_eastmoney` HTTP JSON 解析（mock requests，含空字段、异常包装、北交所短路） - `financial._recent_quarters` 季度滚动跨年 - `market_daily._is_20pct` 主板/创业板/科创板/北交所判定 +- `tdx_blocks.load_infoharbor_blocks` 通达信本地板块缓存解析 +- `sector._fetch_concept_list_ths` / `sector._fetch_concept_stocks_ths` 同花顺概念板块解析 - `sector.fetch_sector` 概念板块增量跳过逻辑 - `src.log.get_logger` 命名空间、handler 幂等、env 控制 level diff --git a/config.example.yaml b/config.example.yaml index 71cb730..c4afdd5 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -15,3 +15,8 @@ fetch: retry: 3 # 并发线程数；BaoStock 单源建议 1；多源轮换（baostock/sina/tencent/eastmoney）可适度提高至 2~4 workers: 1 + +# 通达信本地客户端配置 +tdx: + # 通达信安装目录 + dir: "C:\\new_tdx64" diff --git a/src/config.py b/src/config.py index 119cc5d..176c174 100644 --- a/src/config.py +++ b/src/config.py @@ -45,3 +45,17 @@ def get_fetch_config(config: dict | None = None) -> dict: if config is None: config = load_config() return config.get("fetch", {"delay": 0.5, "retry": 3}) + + +def get_tdx_config(config: dict | None = None) -> dict: + """返回通达信相关配置。""" + if config is None: + config = load_config() + return config.get("tdx", {}) + + +def get_tdx_dir(config: dict | None = None) -> str | None: + """返回通达信安装目录。""" + tdx_cfg = get_tdx_config(config) + tdx_dir = tdx_cfg.get("dir") + return str(tdx_dir) if tdx_dir else None diff --git a/src/db.py b/src/db.py index 6e0e52c..e6726eb 100644 --- a/src/db.py +++ b/src/db.py @@ -8,7 +8,7 @@ - stock_dividend: 分红送转 - trading_day: 交易日历（用于判断数据完整性） - stock_no_data: 无数据/停牌记录（避免重复抓取） - - stock_min5/15/30/60: 分钟K线（4 张分表） + - stock_min5: 分钟K线 - index_daily: 主要指数日线 - market_daily: 每日涨跌停统计（10%/20% 板块分别计数） - stock_sector: 行业 + 地域分类 @@ -185,63 +185,6 @@ class StockMin5(Base): amount = Column(Float, comment="成交额") -# ── 15分钟K线 ── -class StockMin15(Base): - __tablename__ = "stock_min15" - __table_args__ = ( - UniqueConstraint("code", "datetime", name="uq_min15_code_dt"), - Index("ix_min15_datetime", "datetime"), - ) - - id = Column(Integer, primary_key=True, autoincrement=True) - code = Column(String(10), nullable=False, comment="股票代码") - datetime = Column(DateTime, nullable=False, comment="时间") - open = Column(Float, comment="开盘价") - high = Column(Float, comment="最高价") - low = Column(Float, comment="最低价") - close = Column(Float, comment="收盘价") - volume = Column(Float, comment="成交量") - amount = Column(Float, comment="成交额") - - -# ── 30分钟K线 ── -class StockMin30(Base): - __tablename__ = "stock_min30" - __table_args__ = ( - UniqueConstraint("code", "datetime", name="uq_min30_code_dt"), - Index("ix_min30_datetime", "datetime"), - ) - - id = Column(Integer, primary_key=True, autoincrement=True) - code = Column(String(10), nullable=False, comment="股票代码") - datetime = Column(DateTime, nullable=False, comment="时间") - open = Column(Float, comment="开盘价") - high = Column(Float, comment="最高价") - low = Column(Float, comment="最低价") - close = Column(Float, comment="收盘价") - volume = Column(Float, comment="成交量") - amount = Column(Float, comment="成交额") - - -# ── 60分钟K线 ── -class StockMin60(Base): - __tablename__ = "stock_min60" - __table_args__ = ( - UniqueConstraint("code", "datetime", name="uq_min60_code_dt"), - Index("ix_min60_datetime", "datetime"), - ) - - id = Column(Integer, primary_key=True, autoincrement=True) - code = Column(String(10), nullable=False, comment="股票代码") - datetime = Column(DateTime, nullable=False, comment="时间") - open = Column(Float, comment="开盘价") - high = Column(Float, comment="最高价") - low = Column(Float, comment="最低价") - close = Column(Float, comment="收盘价") - volume = Column(Float, comment="成交量") - amount = Column(Float, comment="成交额") - - # ── 每日涨跌停统计 ── class MarketDaily(Base): __tablename__ = "market_daily" diff --git a/src/fetchers/intraday.py b/src/fetchers/intraday.py index bb40b79..a1812d7 100644 --- a/src/fetchers/intraday.py +++ b/src/fetchers/intraday.py @@ -1,37 +1,39 @@ -"""分钟K线抓取模块 — 使用 BaoStock +"""分钟K线抓取模块 — BaoStock / 通达信本地客户端 -BaoStock 支持 5/15/30/60 分钟K线，分别存入四张表。 -注意：BaoStock 不支持 1 分钟线，且不含北交所。 +默认使用通达信本地客户端。 +如果临时需要，也可以显式切换回 BaoStock。 用法： python -m src.main --intraday --freq 5 # 5分钟K线 - python -m src.main --intraday --freq all # 全部频率 python -m src.main --intraday --start-date 20260508 --end-date 20260509 - python -m src.main --intraday --symbol 000001 --freq 30 + python -m src.main --intraday --symbol 000001 """ import time +import random from datetime import datetime, timedelta + +import pandas as pd import baostock as bs from src.baostock_conn import bs_query, code_to_bs, bs_login from src.config import get_fetch_config -from src.db import ( - StockMin5, StockMin15, StockMin30, StockMin60, - TradingDay, batch_upsert, get_session, get_stock_codes, -) +from src.db import StockMin5, TradingDay, batch_upsert, get_session, get_stock_codes +from src.fetchers.tdx_client import get_market_data as tdx_get_market_data +from src.fetchers.tdx_client import format_tdx_code_for_log +from src.fetchers.tdx_client import normalize_tdx_code +from src.fetchers.tdx_client import refresh_minute_cache as tdx_refresh_minute_cache from src.log import get_logger from sqlalchemy import select, func, distinct, text _logger = get_logger("intraday") -VALID_FREQ = ("5", "15", "30", "60") +VALID_FREQ = ("5",) + +TDX_AUTOCACHE_PERIODS = ("1m", "5m") FREQ_MODEL = { "5": StockMin5, - "15": StockMin15, - "30": StockMin30, - "60": StockMin60, } @@ -54,6 +56,110 @@ def _parse_datetime(date_str: str, time_str: str) -> str | None: return f"{date_str} {time_str}" +def _parse_tdx_datetime(value) -> datetime: + """将通达信返回的时间索引转成 `datetime`。""" + if isinstance(value, datetime): + return value + if hasattr(value, "to_pydatetime"): + return value.to_pydatetime() + return pd.to_datetime(value).to_pydatetime() + + +def _tdx_market_data_to_rows(code: str, market_data: dict) -> list[dict]: + """把通达信 `get_market_data` 的返回值整理成数据库行。""" + if not market_data: + return [] + + lower_map = {str(key).lower(): value for key, value in market_data.items()} + field_map = { + "open": "open", + "high": "high", + "low": "low", + "close": "close", + "volume": "volume", + "amount": "amount", + } + + base_frame = None + for field in ("close", "open", "high", "low", "volume", "amount"): + frame = lower_map.get(field) + if hasattr(frame, "index") and len(frame.index) > 0: + base_frame = frame + break + + if base_frame is None: + return [] + + rows: list[dict] = [] + for ts in base_frame.index: + row = { + "code": code, + "datetime": _parse_tdx_datetime(ts), + } + valid_value = False + for field, column in field_map.items(): + frame = lower_map.get(field) + if frame is None or ts not in frame.index: + continue + cell = frame.loc[ts] + if isinstance(cell, pd.Series): + if code in cell.index: + value = cell[code] + else: + value = cell.iloc[0] + else: + value = cell + value = _clean(value) + row[column] = value + if value is not None: + valid_value = True + if valid_value: + rows.append(row) + + return rows + + +def _tdx_fetch_rows_with_autocache( + code: str, + tdx_code: str, + tdx_period: str, + start_time: str, + end_time: str, +) -> tuple[list[dict], bool]: + """先读取通达信本地数据，缺缓存时自动刷新一次后重试。""" + def _fetch_rows(period: str) -> list[dict]: + market_data = tdx_get_market_data( + [tdx_code], + period=period, + start_time=start_time, + end_time=end_time, + count=-1, + dividend_type="none", + fill_data=True, + ) + return _tdx_market_data_to_rows(code, market_data) + + rows = _fetch_rows(tdx_period) + if rows: + return rows, False + + _logger.info( + "通达信 %s(%s) 首次没有返回数据，自动刷新本地分钟缓存后重试", + code, tdx_code, + ) + tdx_refresh_minute_cache( + [tdx_code], + periods=TDX_AUTOCACHE_PERIODS, + batch_size=1, + pause_seconds=0, + ) + rows = _fetch_rows(tdx_period) + if rows: + return rows, True + + return [], True + + def _get_intraday_gaps(model, code: str, sd: str, ed: str) -> list[tuple[str, str]]: """分析单只股票在 [sd, ed] 范围内的分钟K线缺口，返回缺失区间列表""" session = get_session() @@ -98,32 +204,35 @@ def _get_intraday_gaps(model, code: str, sd: str, ed: str) -> list[tuple[str, st def fetch_intraday(start_date: str | None = None, end_date: str | None = None, - symbol: str | None = None, freq: str = "5"): + symbol: str | None = None, freq: str = "5", + source: str = "tdx", prewarm_cache: bool = True): """抓取分钟K线行情 Args: start_date: 开始日期 YYYYMMDD，默认30天前 end_date: 结束日期 YYYYMMDD，默认今天 symbol: 单只股票代码，默认全部 - freq: K线频率 5/15/30/60/all + freq: K线频率 5 + source: 数据源，`tdx` 或 `baostock` + prewarm_cache: 使用通达信时，是否先批量刷新本地分钟缓存 """ - if freq == "all": - freqs = list(VALID_FREQ) - elif freq in VALID_FREQ: - freqs = [freq] - else: - _logger.error("不支持的频率 %s，可选: %s, all", freq, ", ".join(VALID_FREQ)) + if freq not in VALID_FREQ: + _logger.error("不支持的频率 %s，可选: %s", freq, ", ".join(VALID_FREQ)) return - for f in freqs: - _fetch_one_freq(f, start_date, end_date, symbol) + if source == "tdx": + _fetch_one_freq_tdx("5", start_date, end_date, symbol, prewarm_cache=prewarm_cache) + else: + _fetch_one_freq_baostock("5", start_date, end_date, symbol) -def _fetch_one_freq(freq: str, start_date: str | None, end_date: str | None, - symbol: str | None): +def _fetch_one_freq_baostock(freq: str, start_date: str | None, end_date: str | None, + symbol: str | None): """抓取单个频率的分钟K线""" cfg = get_fetch_config() - delay = cfg.get("delay", 0.1) + base_delay = float(cfg.get("delay", 0.1)) + # 分钟K线比日线更容易触发限流，默认放慢到 1 秒以上。 + delay = max(base_delay, 1.0) model = FREQ_MODEL[freq] if end_date is None: @@ -200,6 +309,7 @@ def _fetch_one_freq(freq: str, start_date: str | None, end_date: str | None, if rows: batch_upsert(model, rows, ["code", "datetime"]) total_rows += len(rows) + time.sleep(delay + random.uniform(0, delay * 0.2)) if total_rows: success += 1 except Exception: @@ -212,11 +322,125 @@ def _fetch_one_freq(freq: str, start_date: str | None, end_date: str | None, i+1, total, success, fail, elapsed, ) - if not symbol: - time.sleep(delay) - total_time = time.time() - t_start _logger.info( "%s分钟K线抓取完成，成功:%d 失败:%d 总耗时:%.1fs", freq, success, fail, total_time, ) + + +def _fetch_one_freq_tdx(freq: str, start_date: str | None, end_date: str | None, + symbol: str | None, prewarm_cache: bool = True): + """使用通达信本地客户端抓取单个频率的分钟K线。""" + cfg = get_fetch_config() + base_delay = float(cfg.get("delay", 0.1)) + delay = max(base_delay, 0.2) + model = FREQ_MODEL[freq] + tdx_period = "5m" + + if end_date is None: + end_date = datetime.now().strftime("%Y%m%d") + if start_date is None: + start_date = (datetime.now() - timedelta(days=30)).strftime("%Y%m%d") + + sd = f"{start_date[:4]}-{start_date[4:6]}-{start_date[6:8]}" + ed = f"{end_date[:4]}-{end_date[4:6]}-{end_date[6:8]}" + + if symbol: + codes = [symbol] + else: + codes = get_stock_codes() + if not codes: + _logger.error("无股票列表，请先运行 --stock-info") + return + + need_fetch = [] + if symbol: + need_fetch = [(symbol, [(sd, ed)])] + else: + skip = 0 + for code in codes: + gaps = _get_intraday_gaps(model, code, sd, ed) + if gaps: + need_fetch.append((code, gaps)) + else: + skip += 1 + codes = [c for c, _ in need_fetch] + + if not need_fetch: + _logger.info("通达信 %s分钟K线 %s ~ %s 数据已完整，跳过", freq, sd, ed) + return + + skip_msg = f"（跳过 {skip} 只已完整）" if not symbol and skip else "" + preview = ", ".join(format_tdx_code_for_log(code) for code in codes[:8]) + if len(codes) > 8: + preview += f", ...(+{len(codes) - 8})" + _logger.info( + "正在使用通达信抓取%s分钟K线 %s ~ %s，需补缺 %d 只%s，目标: [%s]", + freq, sd, ed, len(need_fetch), skip_msg, preview, + ) + + # 全市场抓取先批量预热，单只股票则走按需自动缓存，避免重复刷新。 + if not symbol and prewarm_cache: + try: + refresh_results = tdx_refresh_minute_cache(codes, periods=TDX_AUTOCACHE_PERIODS) + refresh_failed = [item for item in refresh_results if item.get("result", {}).get("ErrorId") not in (None, "0", 0)] + if refresh_failed: + _logger.warning( + "通达信分钟缓存刷新后仍有 %d 个批次失败，分钟线可能不完整", + len(refresh_failed), + ) + except Exception as exc: + _logger.warning("通达信分钟缓存刷新失败，继续尝试取数：%s", exc) + + total = len(need_fetch) + success = 0 + fail = 0 + t_start = time.time() + + for i, (code, gaps) in enumerate(need_fetch): + total_rows = 0 + tdx_code = normalize_tdx_code(code) + try: + for gap_sd, gap_ed in gaps: + start_time = f"{gap_sd.replace('-', '')}000000" + end_time = f"{gap_ed.replace('-', '')}235959" + rows, _refreshed = _tdx_fetch_rows_with_autocache( + code, + tdx_code, + tdx_period, + start_time, + end_time, + ) + if rows: + batch_upsert(model, rows, ["code", "datetime"]) + total_rows += len(rows) + time.sleep(delay + random.uniform(0, delay * 0.2)) + if total_rows: + success += 1 + else: + fail += 1 + _logger.warning( + "通达信 %s分钟K线 %s(%s) 没有返回数据，区间 %s ~ %s，period=%s；" + "通常表示本地客户端没有下载该周期的分钟缓存，或该股在此区间无分钟数据", + freq, code, tdx_code, sd, ed, tdx_period, + ) + except Exception as exc: + fail += 1 + _logger.warning( + "通达信 %s分钟K线 %s(%s) 失败: %s，区间 %s ~ %s，period=%s", + freq, code, tdx_code, exc, sd, ed, tdx_period, + ) + + if (i + 1) % 100 == 0: + elapsed = time.time() - t_start + _logger.info( + "[%d/%d] 进度... 成功:%d 失败:%d 已用时:%.0fs", + i + 1, total, success, fail, elapsed, + ) + + total_time = time.time() - t_start + _logger.info( + "通达信 %s分钟K线抓取完成，成功:%d 失败:%d 总耗时:%.1fs", + freq, success, fail, total_time, + ) diff --git a/src/fetchers/sector.py b/src/fetchers/sector.py index 54ea048..9a9e67d 100644 --- a/src/fetchers/sector.py +++ b/src/fetchers/sector.py @@ -1,98 +1,146 @@ -"""概念板块数据抓取 — 东方财富 +"""概念板块数据抓取 — 通达信本地 + 同花顺 数据源： - - 概念板块列表：东方财富 push2 API - - 成分股：每个概念的成分股代码列表 + - 优先：通达信本地概念板块缓存 + - 备用：同花顺概念板块页面 用法： python -m src.main --sector """ import time +from contextlib import redirect_stderr, redirect_stdout +from io import StringIO + +import pandas as pd import requests +from bs4 import BeautifulSoup from src.db import StockConcept, batch_upsert, get_session +from src.fetchers.tdx_blocks import load_infoharbor_blocks from src.log import get_logger -from sqlalchemy import select, func +from sqlalchemy import func, select _logger = get_logger("sector") -_HEADERS = { - "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", -} +def _is_a_stock_code(code: str) -> bool: + return len(code) == 6 and code.isdigit() and not code.startswith("920") -_EM_CONCEPT_LIST_URL = "https://push2.eastmoney.com/api/qt/clist/get" -_EM_CONCEPT_STOCKS_URL = "https://push2.eastmoney.com/api/qt/clist/get" + +def _fetch_concept_list_ths() -> list[dict]: + """使用同花顺获取全部概念板块列表。""" + try: + import akshare as ak + + # akshare 内部会打印进度条，这里静默掉，避免污染主流程日志。 + with redirect_stdout(StringIO()), redirect_stderr(StringIO()): + df = ak.stock_board_concept_name_ths() + if df is None or df.empty: + return [] + rows: list[dict] = [] + for _, row in df.iterrows(): + code = str(row.get("code", "")).strip() + name = str(row.get("name", "")).strip() + if code and name: + rows.append({"code": code, "name": name}) + return rows + except Exception as e: + _logger.error("同花顺 获取概念板块列表失败: %s", e) + return [] def _fetch_concept_list() -> list[dict]: - """从东方财富获取全部概念板块列表（分页+重试）""" - all_items = [] - page = 1 - max_retry = 3 - while True: - params = { - "pn": page, "pz": 200, "po": 1, "np": 1, - "ut": "bd1d9ddb04089700cf9c27f6f7426281", - "fltt": 2, "invt": 2, - "fid": "f3", - "fs": "m:90+t:3", - "fields": "f2,f3,f12,f14", - "_": int(time.time() * 1000), - } - for attempt in range(1, max_retry + 1): + """获取全部概念板块列表。""" + concepts = _fetch_concept_list_tdx() + if concepts: + return concepts + return _fetch_concept_list_ths() + + +def _fetch_concept_list_tdx() -> list[dict]: + """使用通达信本地概念板块缓存获取全部板块。""" + blocks = load_infoharbor_blocks() + return [{"code": block.code, "name": block.name} for block in blocks if block.stocks] + + +def _extract_ths_stock_codes(html: str) -> set[str]: + """从同花顺概念详情页表格中提取 A 股代码。""" + codes: set[str] = set() + try: + tables = pd.read_html(StringIO(html)) + except ValueError: + return codes + + for df in tables: + if "代码" not in df.columns: + continue + for value in df["代码"].tolist(): + code = str(value).strip().zfill(6) + if _is_a_stock_code(code): + codes.add(code) + return codes + + +def _fetch_concept_stocks_ths(concept_code: str) -> list[str]: + """使用同花顺概念详情页获取单个概念板块成分股。""" + headers = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", + "Referer": "https://q.10jqka.com.cn/gn/", + } + base_url = f"https://q.10jqka.com.cn/gn/detail/code/{concept_code}/" + + try: + r = requests.get(base_url, headers=headers, timeout=15) + r.encoding = "gbk" + soup = BeautifulSoup(r.text, features="lxml") + page_info = soup.find(name="span", attrs={"class": "page_info"}) + page_count = 1 + if page_info and "/" in page_info.text: try: - r = requests.get(_EM_CONCEPT_LIST_URL, params=params, headers=_HEADERS, timeout=15) - data = r.json().get("data", {}) or {} - items = data.get("diff", []) or [] - total = data.get("total", 0) - break + page_count = int(page_info.text.strip().split("/")[-1]) + except ValueError: + page_count = 1 + + codes = _extract_ths_stock_codes(r.text) + for page in range(2, page_count + 1): + page_url = f"{base_url}page/{page}/" + try: + page_resp = requests.get(page_url, headers=headers, timeout=15) + page_resp.encoding = "gbk" + codes.update(_extract_ths_stock_codes(page_resp.text)) except Exception as e: - if attempt < max_retry: - _logger.warning("获取概念列表第 %d 页第 %d 次失败，%ds 后重试: %s", - page, attempt, attempt * 2, e) - time.sleep(attempt * 2) - else: - _logger.error("获取概念板块列表第 %d 页失败（已重试 %d 次）: %s", page, max_retry, e) - items = [] - total = 0 - - for item in items: - if item.get("f12") and item.get("f14"): - all_items.append({"code": item["f12"], "name": item["f14"]}) - - if not items or len(all_items) >= total: - break - page += 1 - time.sleep(0.1) - - return all_items + _logger.warning("同花顺 获取概念 %s 第 %d 页失败: %s", concept_code, page, e) + time.sleep(0.03) + return sorted(codes) + except Exception as e: + _logger.warning("同花顺 获取概念 %s 成分股失败: %s", concept_code, e) + return [] def _fetch_concept_stocks(concept_code: str) -> list[str]: """获取单个概念板块的成分股代码列表（仅保留 A 股）""" - params = { - "pn": 1, "pz": 5000, "po": 1, "np": 1, - "ut": "bd1d9ddb04089700cf9c27f6f7426281", - "fltt": 2, "invt": 2, - "fid": "f3", - "fs": f"b:{concept_code}+f:!50", - "fields": "f12", - "_": int(time.time() * 1000), - } - try: - r = requests.get(_EM_CONCEPT_STOCKS_URL, params=params, headers=_HEADERS, timeout=15) - data = r.json().get("data", {}) or {} - items = data.get("diff", []) or [] - return [item["f12"] for item in items - if item.get("f12") and len(item["f12"]) == 6 and item["f12"].isdigit()] - except Exception: - return [] + stocks = _fetch_concept_stocks_tdx(concept_code) + if stocks: + return stocks + return _fetch_concept_stocks_ths(concept_code) + + +def _fetch_concept_stocks_tdx(concept_code: str) -> list[str]: + """使用通达信本地概念板块缓存获取成分股。""" + for block in load_infoharbor_blocks(): + if concept_code == block.code or concept_code == block.name: + return list(block.stocks) + return [] def fetch_sector(): """抓取全部概念板块及成分股，写入 stock_concept 表""" - # 检查已有概念数据 + _logger.info("正在获取概念板块列表...") + concepts = _fetch_concept_list() + if not concepts: + _logger.warning("概念板块列表获取失败") + return + session = get_session() try: existing_count = session.execute( @@ -104,16 +152,10 @@ def fetch_sector(): finally: session.close() - if existing_concepts >= 450: + if existing_concepts >= len(concepts): _logger.info("概念板块已有 %d 个概念、%d 条记录，跳过", existing_concepts, existing_count) return - _logger.info("正在获取概念板块列表...") - concepts = _fetch_concept_list() - if not concepts: - _logger.warning("概念板块列表获取失败") - return - _logger.info("共 %d 个概念板块，开始抓取成分股...", len(concepts)) t_start = time.time() rows = [] @@ -127,14 +169,14 @@ def fetch_sector(): }) if (i + 1) % 50 == 0: elapsed = time.time() - t_start - _logger.info("[%d/%d] 已处理 耗时:%.0fs", i+1, len(concepts), elapsed) + _logger.info("[%d/%d] 已处理 耗时:%.0fs", i + 1, len(concepts), elapsed) time.sleep(0.05) if rows: batch_upsert(StockConcept, rows, ["code", "concept_code"]) _logger.info( "概念板块写入完成，%d 个概念，%d 条记录，耗时:%.0fs", - len(concepts), len(rows), time.time()-t_start, + len(concepts), len(rows), time.time() - t_start, ) else: _logger.warning("无概念板块数据") diff --git a/src/main.py b/src/main.py index 7641def..1d36d1a 100644 --- a/src/main.py +++ b/src/main.py @@ -6,10 +6,12 @@ python -m src.main --daily --start-date 20260501 --end-date 20260508 python -m src.main --financial --symbol 000001 python -m src.main --dividend - python -m src.main --intraday # 默认5分钟，近30天 - python -m src.main --intraday --freq all # 全部频率 + python -m src.main --intraday # 默认5分钟，近30天，通达信本地 python -m src.main --intraday --start-date 20260508 --end-date 20260509 - python -m src.main --intraday --symbol 000001 --freq 30 + python -m src.main --intraday --intraday-source tdx # 使用通达信本地客户端 + python -m src.main --tdx-cache # 全历史回填通达信5分钟数据 + python -m src.main --tdx-local-cache # 预热通达信本地分钟缓存并自动校验，不写数据库 + python -m src.main --tdx-verify-cache # 只验证通达信本地分钟缓存是否可读 python -m src.main --sector # 概念板块及成分股 python -m src.main --index # 指数日线（上证/沪深300/创业板等） python -m src.main --market-daily # 汇总每日涨跌停统计（依赖 stock_daily） @@ -19,11 +21,177 @@ import argparse from src.baostock_conn import bs_login, bs_logout from src.config import load_config -from src.db import init_db +from src.db import get_ipo_dates, init_db from src.log import get_logger _logger = get_logger("main") +TDX_LOCAL_CACHE_PERIODS = ("1m", "5m") + + +def _resolve_tdx_backfill_start_date(explicit_start_date: str | None, symbol: str | None) -> str: + """为通达信分钟回填选择更合理的起始日期。""" + if explicit_start_date: + return explicit_start_date + + ipo_dates = get_ipo_dates() + if symbol: + ipo_date = ipo_dates.get(symbol) + if ipo_date: + return ipo_date.replace("-", "") + return "19900101" + + if ipo_dates: + return min(ipo_dates.values()).replace("-", "") + return "19900101" + + +def _format_cache_preview(results: list[dict], *, limit: int = 10) -> tuple[int, int, str]: + """把缓存验证结果整理成更直观的摘要。""" + from src.fetchers.tdx_client import format_tdx_code_for_log + + hit_items = [item for item in results if item.get("has_data")] + miss_items = [item for item in results if not item.get("has_data")] + + if miss_items: + preview = ", ".join( + format_tdx_code_for_log(item["code"]) for item in miss_items[:limit] + ) + if len(miss_items) > limit: + preview += f", ...(+{len(miss_items) - limit})" + else: + preview = "" + + return len(hit_items), len(miss_items), preview + + +def _log_cache_summary( + *, + title: str, + results: list[dict], + single_symbol: bool = False, +) -> None: + """输出更直观的缓存结论。""" + if not results: + _logger.warning("%s：没有拿到任何验证结果", title) + return + + hit, miss, preview = _format_cache_preview(results) + total = len(results) + + if single_symbol: + item = results[0] + status = "可读" if item.get("has_data") else "不可读" + from src.fetchers.tdx_client import format_tdx_code_for_log + + _logger.info( + "%s：%s %s，行数 %d", + title, + status, + format_tdx_code_for_log(item["code"]), + item.get("row_count", 0), + ) + return + + if miss: + _logger.warning( + "%s：已可读 %d/%d，只缺失 %d 只%s%s", + title, + hit, + total, + miss, + "，缺失名单: " if preview else "", + preview, + ) + else: + _logger.info("%s：全部 %d 只可读", title, total) + + +def _run_tdx_local_cache(symbol: str | None) -> None: + """只预热通达信本地缓存，并在结束后自动校验可读性。""" + from src.db import get_stock_codes + from src.fetchers.tdx_client import ( + format_tdx_code_for_log, + refresh_minute_cache, + verify_minute_cache, + ) + + period_label = ",".join(TDX_LOCAL_CACHE_PERIODS) + if symbol: + _logger.info( + "开始预热通达信本地分钟缓存 %s，周期 %s，不写数据库", + format_tdx_code_for_log(symbol), + period_label, + ) + codes = [symbol] + else: + try: + codes = get_stock_codes() + except Exception as exc: + _logger.error("读取 stock_info 失败，无法预热全量通达信本地缓存：%s", exc) + return + + if not codes: + _logger.error("没有可用股票代码，请先运行 --stock-info") + return + + _logger.info( + "开始预热通达信全市场本地分钟缓存，股票数: %d，周期 %s，不写数据库", + len(codes), + period_label, + ) + refresh_minute_cache(codes, periods=TDX_LOCAL_CACHE_PERIODS) + _logger.info("通达信本地分钟缓存预热完成，开始自动校验") + results = verify_minute_cache(codes) + _log_cache_summary( + title="通达信本地缓存自动校验结果", + results=results, + single_symbol=bool(symbol), + ) + + +def _run_tdx_verify_cache( + start_date: str | None, + end_date: str | None, + symbol: str | None, +) -> None: + """只验证通达信本地分钟缓存是否可读。""" + from src.db import get_stock_codes + from src.fetchers.tdx_client import format_tdx_code_for_log, verify_minute_cache + + if symbol: + codes = [symbol] + _logger.info( + "开始验证通达信本地分钟缓存 %s，区间 %s ~ %s", + format_tdx_code_for_log(symbol), + start_date or "默认近30天", + end_date or "today", + ) + else: + try: + codes = get_stock_codes() + except Exception as exc: + _logger.error("读取 stock_info 失败，无法验证全量通达信本地缓存：%s", exc) + return + + if not codes: + _logger.error("没有可用股票代码，请先运行 --stock-info") + return + + _logger.info( + "开始验证通达信全市场本地分钟缓存，股票数: %d，区间 %s ~ %s", + len(codes), + start_date or "默认近30天", + end_date or "today", + ) + + results = verify_minute_cache(codes, start_date=start_date, end_date=end_date) + _log_cache_summary( + title="通达信本地缓存验证结果", + results=results, + single_symbol=bool(symbol), + ) + def main(): parser = argparse.ArgumentParser(description="A股数据抓取工具（BaoStock + 多源容灾）") @@ -37,8 +205,17 @@ def main(): parser.add_argument("--dividend", action="store_true", help="抓取分红送转") parser.add_argument("--intraday", action="store_true", help="抓取分钟K线行情") parser.add_argument("--freq", type=str, default="5", - choices=["5", "15", "30", "60", "all"], - help="分钟K线频率（默认5，all=全部）") + choices=["5"], + help="分钟K线频率（当前仅支持5分钟）") + parser.add_argument("--intraday-source", type=str, default="tdx", + choices=["baostock", "tdx"], + help="分钟K线数据源（默认tdx，baostock=备用来源）") + parser.add_argument("--tdx-cache", action="store_true", + help="全历史回填通达信5分钟数据（先预热本地缓存，再写入分钟表）") + parser.add_argument("--tdx-local-cache", action="store_true", + help="预热通达信本地分钟缓存并自动校验，不写数据库") + parser.add_argument("--tdx-verify-cache", action="store_true", + help="只验证通达信本地分钟缓存是否可读，不写数据库") parser.add_argument("--sector", action="store_true", help="抓取概念板块及成分股") parser.add_argument("--index", action="store_true", help="抓取指数日线行情") parser.add_argument("--market-daily", action="store_true", @@ -51,11 +228,32 @@ def main(): if not any([args.stock_info, args.trading_day, args.daily, args.financial, args.dividend, args.intraday, - args.sector, args.index, args.market_daily]): + args.sector, args.index, args.market_daily, args.tdx_cache, + args.tdx_local_cache, args.tdx_verify_cache]): parser.print_help() return load_config() + + if args.tdx_verify_cache: + if any([args.stock_info, args.trading_day, args.daily, + args.financial, args.dividend, args.intraday, + args.sector, args.index, args.market_daily, args.tdx_cache, + args.tdx_local_cache]): + _logger.warning("`--tdx-verify-cache` 为纯验证命令，已忽略其它任务参数") + _run_tdx_verify_cache(args.start_date, args.end_date, args.symbol) + _logger.info("全部任务完成") + return + + if args.tdx_local_cache: + if any([args.stock_info, args.trading_day, args.daily, + args.financial, args.dividend, args.intraday, + args.sector, args.index, args.market_daily, args.tdx_cache]): + _logger.warning("`--tdx-local-cache` 为纯本地缓存命令，已忽略其它任务参数") + _run_tdx_local_cache(args.symbol) + _logger.info("全部任务完成") + return + init_db() # market_daily 只读 stock_daily，无需 BaoStock 登录，提前处理 @@ -65,7 +263,7 @@ def main(): # 若仅运行 market-daily，避免无谓的登录 if not any([args.stock_info, args.trading_day, args.daily, args.financial, args.dividend, args.intraday, - args.sector, args.index]): + args.sector, args.index, args.tdx_cache]): _logger.info("全部任务完成") return @@ -75,11 +273,65 @@ def main(): fetch_sector() if not any([args.stock_info, args.trading_day, args.daily, args.financial, args.dividend, args.intraday, - args.index, args.market_daily]): + args.index, args.market_daily, args.tdx_cache]): _logger.info("全部任务完成") return - bs_login() + if args.tdx_cache: + from src.fetchers.intraday import fetch_intraday + from src.db import get_stock_codes + from src.fetchers.tdx_client import refresh_all_minute_cache + + start_date = _resolve_tdx_backfill_start_date(args.start_date, args.symbol) + end_date = args.end_date + if args.symbol: + _logger.info( + "开始回填指定股票分钟数据 %s，区间 %s ~ %s，频率 5", + args.symbol, + start_date, + end_date or "today", + ) + fetch_intraday( + start_date=start_date, + end_date=end_date, + symbol=args.symbol, + freq="5", + source="tdx", + prewarm_cache=True, + ) + else: + codes = get_stock_codes() + if not codes: + _logger.error("没有可用股票代码，请先运行 --stock-info") + else: + _logger.info( + "开始回填通达信全市场分钟数据，股票数: %d，区间 %s ~ %s，频率 5", + len(codes), + start_date, + end_date or "today", + ) + refresh_all_minute_cache(stock_list=codes) + fetch_intraday( + start_date=start_date, + end_date=end_date, + freq="5", + source="tdx", + prewarm_cache=False, + ) + + if not any([args.stock_info, args.trading_day, args.daily, + args.financial, args.dividend, args.intraday, + args.sector, args.index, args.market_daily]): + _logger.info("全部任务完成") + return + + need_bs_login = any([args.stock_info, args.trading_day, args.daily, + args.financial, args.dividend, args.index]) or ( + args.intraday and args.intraday_source != "tdx" + ) + + if need_bs_login: + bs_login() try: if args.stock_info: from src.fetchers.stock_list import fetch_stock_list @@ -105,7 +357,8 @@ def main(): if args.intraday: from src.fetchers.intraday import fetch_intraday fetch_intraday(start_date=args.start_date, end_date=args.end_date, - symbol=args.symbol, freq=args.freq) + symbol=args.symbol, freq=args.freq, + source=args.intraday_source) if args.index: from src.fetchers.index import fetch_index @@ -113,7 +366,8 @@ def main(): _logger.info("全部任务完成") finally: - bs_logout() + if need_bs_login: + bs_logout() if __name__ == "__main__": diff --git a/tests/README.md b/tests/README.md index 8de253c..2873dc1 100644 --- a/tests/README.md +++ b/tests/README.md @@ -9,7 +9,7 @@ pip install -e .[dev] # 或 pip install pytest pytest -v ``` -## 覆盖范围（43 个用例） +## 覆盖范围（48 个用例） - `test_code_mapping.py` —— `baostock_conn.code_to_bs` 代码前缀映射（5） - `test_daily_derived.py` —— `daily._fill_derived_fields` 振幅/涨跌幅补算（5） @@ -18,4 +18,6 @@ pytest -v - `test_financial_quarters.py` —— `financial._recent_quarters` 季度滚动（5） - `test_log.py` —— `log.get_logger` 命名空间、handler 幂等、env 控制 level（5） - `test_market_classification.py` —— `market_daily._is_20pct` 板块判定（4） -- `test_sector_merge.py` —— `sector.fetch_sector` 三种 only 模式合并保护（3） +- `test_sector_http.py` —— 通达信/同花顺概念板块名称/成分股解析（4） +- `test_sector_merge.py` —— `sector.fetch_sector` 通达信增量跳过保护（2） +- `test_tdx_blocks.py` —— 通达信本地板块文件解析（2） diff --git a/tests/test_sector_merge.py b/tests/test_sector_merge.py index 35a90ed..9631217 100644 --- a/tests/test_sector_merge.py +++ b/tests/test_sector_merge.py @@ -6,23 +6,24 @@ from src.fetchers import sector def test_concept_skips_when_enough_data(): - """已有 >= 450 个概念时，跳过抓取""" + """已有数量达到当前板块数量时，跳过抓取。""" mock_session = MagicMock() mock_session.execute.side_effect = [MagicMock(scalar=MagicMock(return_value=150000)), MagicMock(scalar=MagicMock(return_value=460))] with patch.object(sector, "get_session", return_value=mock_session), \ - patch.object(sector, "_fetch_concept_list") as mock_list: + patch.object(sector, "_fetch_concept_list", return_value=[{"code": "880001", "name": "测试板块"}]) as mock_list: sector.fetch_sector() - mock_list.assert_not_called() + mock_list.assert_called_once_with() def test_concept_fetches_when_insufficient(): - """概念数 < 300 时，执行抓取""" + """板块数量不足时，执行抓取。""" mock_session = MagicMock() mock_session.execute.side_effect = [MagicMock(scalar=MagicMock(return_value=0)), MagicMock(scalar=MagicMock(return_value=0))] with patch.object(sector, "get_session", return_value=mock_session), \ - patch.object(sector, "_fetch_concept_list", return_value=[]): + patch.object(sector, "_fetch_concept_list", return_value=[{"code": "880001", "name": "测试板块"}]), \ + patch.object(sector, "_fetch_concept_stocks", return_value=[]): sector.fetch_sector()