claude_dev
774170ec05
采集层(多源各下): - baostock: 日线全字段全量(baostock_daily_fullmarket) + 15min全市场 + 静态(基础/复权/分红/季频/三表) + 成份股 - akshare: 静态(估值/龙虎榜/大宗/融资融券/北向/指数成分/行业/股本/解禁/业绩预告) - xtdata(miniQMT): build_daily_from_xtdata + daily_update_xtdata 数据补全 P0: - ETF全市场: universe 扩展 沪深A股∪ETF∪基金(7414), dividend_type='front' 前复权 - 历史成份股(治幸存者偏差): index_const_hist_download 深证/国证 adjust_cni 4指数 + 中证1000/2000快照 + 新浪交叉校验 - 退市K线: baostock_delisted_download + import_delisted_to_db(实证 Day1 fetch_all_stocks 已含退市) 灌库: - import_baostock_to_db: daily_baostock_full(5537股/1826万行,18字段)+ bs_index_constituent + bs_adjust_factor - INSERT OR REPLACE 幂等, WAL+busy_timeout, dbbardata 不碰 每日增量 #7(用户决策A: VPS直跑): - daily_update_static: login探针防黑名单graceful skip + LOOKBACK7 + query_stock_basic含退市 + INSERT OR REPLACE + QUERY_COUNT守48000/天 设计文档: spec(13节三层融合) + P0 plan + 数据gap设计
984 lines
35 KiB
Python
984 lines
35 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""AKShare A 股静态数据全量下载 (16 类, 四种模式, 只产 parquet)
|
||
|
||
目标:
|
||
在 Windows Server 2022 VPS (49.232.102.198) 上用 akshare 全量拉取 A 股
|
||
静态数据 (估值/北向/股本/十大股东/三大报表/财务摘要/龙虎榜/大宗/融资融券/
|
||
解禁/业绩预告/业绩快报/指数成分/申万行业), 输出 import-ready parquet.
|
||
|
||
四种模式:
|
||
- 模式A per-stock (5500股循环, 每股1调用) → <code>.<EXC>_<type>.parquet
|
||
- 模式B per-date (交易日循环, 每日1调用) → <date>_<type>.parquet
|
||
- 模式C per-period (报告期循环, 每期1调用) → <period>_<type>.parquet
|
||
- 模式D one-shot (单次调用) → <type>.parquet
|
||
|
||
硬约束 (踩过的坑 / 用户铁律):
|
||
1. 单线程串行, 严禁并发 (东财限流严, 并发会封 IP)
|
||
2. 每次 akshare 调用后 sleep AK_INTERVAL=0.8s (防封)
|
||
3. 每次 akshare 调用 30s 硬超时 (akshare 经常挂死, 用 ThreadPoolExecutor
|
||
单线程 + future.result(timeout=30) 保护, 超时计 failed)
|
||
4. 重试退避: 东财 ConnectionError 常见, 3 次重试, 指数退避 (2s/4s/8s)
|
||
5. 断路器: 连续 30 个单位 (stock/date/period) failed → exit 2
|
||
6. empty vs failed 区分:
|
||
- 空 df (无北向持仓 / 退市 / 节假日无龙虎榜) → status='empty' 中性
|
||
- 异常 / 超时 → status='failed' 计断路器
|
||
7. marker 断点续传 (per-unit): 只在成功写 parquet 后写 marker
|
||
8. 开头 unset proxy (akshare 底层 requests 读 proxy 环境变量)
|
||
|
||
与 baostock_static_download.py 差异:
|
||
- timeout 机制不同: akshare 用 ThreadPoolExecutor + future.result(timeout)
|
||
(baostock 用 socket.setdefaulttimeout, 对 akshare 不够: akshare 内部
|
||
requests 会重试很久, 必须 future-level kill)
|
||
- 四种模式 (baostock 只有 per-stock / per-year / one-shot 三种)
|
||
- symbol 格式适配 code_to_symbol(): 同一 code 在不同端点格式不同
|
||
(stock_value_em="600519", balance_sheet="SH600519",
|
||
top_holders="sh600519", share_change="600519")
|
||
- 重试退避指数 backoff (baostock 重试即重登, akshare 重试即等 2/4/8s)
|
||
|
||
⚠️ 本脚本只依赖 akshare, 不碰 baostock (可与 baostock 15min 下载并行)。
|
||
|
||
用法:
|
||
# 全跑 (16 类)
|
||
python akshare_static_download.py
|
||
# 只跑指定类型 (逗号分隔)
|
||
python akshare_static_download.py --types valuation,index_const
|
||
# 小样测试 (per-stock 类前 3 股)
|
||
python akshare_static_download.py --types valuation --limit 3
|
||
# 指定股票 (6 位 code, 自动猜交易所)
|
||
python akshare_static_download.py --types balance --codes 600519,000001
|
||
# 日期范围 (per-date 类用)
|
||
python akshare_static_download.py --types dragon_tiger --start 20260101 --end 20260715
|
||
# 强制重下
|
||
python akshare_static_download.py --types valuation --force
|
||
|
||
输出目录结构 (OUT_DIR/data/static):
|
||
{OUT_DIR}/
|
||
├── valuation/ (per-stock)
|
||
│ ├── 600519.SH_valuation.parquet
|
||
│ └── .600519.SH_valuation.akshare
|
||
├── balance/ (per-stock, 带交易所前缀调用)
|
||
│ ├── 600519.SH_balance.parquet
|
||
│ └── .600519.SH_balance.akshare
|
||
├── top_holders/ (per-stock × per-period)
|
||
│ ├── 600519.SH_2020930_top_holders.parquet
|
||
│ └── ...
|
||
├── dragon_tiger/ (per-date)
|
||
│ ├── 20260715_dragon_tiger.parquet
|
||
│ └── .20260715_dragon_tiger.akshare
|
||
├── forecast/ (per-period)
|
||
│ ├── 20251231_forecast.parquet
|
||
│ └── ...
|
||
├── index_const/ (one-shot)
|
||
│ ├── index_const.parquet
|
||
│ └── .index_const.akshare
|
||
└── logs/akshare_static_YYYYMMDD_HHMMSS.log
|
||
|
||
退出码: 0=完成, 1=致命错误 (akshare 装载/列表拉取失败), 2=断路器触发 (可重试)
|
||
"""
|
||
|
||
import argparse
|
||
import concurrent.futures
|
||
import datetime
|
||
import json
|
||
import logging
|
||
import os
|
||
import sys
|
||
import time
|
||
from functools import partial
|
||
from pathlib import Path
|
||
from typing import Any, Callable, List, Optional, Tuple
|
||
|
||
# ======================== 硬约束: unset proxy ========================
|
||
# 必须在 import akshare 之前清理 (akshare 底层 requests 读 proxy 环境变量)
|
||
for _k in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY"):
|
||
os.environ.pop(_k, None)
|
||
|
||
# stdout 行缓冲 (SSH-detached / pipe 重定向时也能看到实时进度)
|
||
try:
|
||
sys.stdout.reconfigure(line_buffering=True)
|
||
except (AttributeError, ValueError):
|
||
pass
|
||
|
||
import akshare as ak # noqa: E402
|
||
import pandas as pd # noqa: E402
|
||
|
||
|
||
# ======================== 配置 (环境变量可覆盖) ========================
|
||
|
||
DEFAULT_OUT_DIR = r"C:\sanguo_vnpy_v2\data\static"
|
||
OUT_DIR = Path(os.environ.get("AK_STATIC_OUT_DIR", DEFAULT_OUT_DIR))
|
||
DEFAULT_LOG_DIR = OUT_DIR / "logs"
|
||
_log_dir_env = os.environ.get("AK_STATIC_LOG_DIR")
|
||
LOG_DIR = Path(_log_dir_env) if _log_dir_env else DEFAULT_LOG_DIR
|
||
|
||
START_DATE_DEFAULT = "2020-01-01"
|
||
# 报告期循环范围 (近5年×4季, forecast/express/top_holders 用).
|
||
# top_holders per-stock × per-period 会有股票×20期组合.
|
||
REPORT_PERIODS = [
|
||
f"{y}{m:02d}{d:02d}"
|
||
for y in range(2020, datetime.date.today().year + 1)
|
||
for (m, d) in [(3, 31), (6, 30), (9, 30), (12, 31)]
|
||
if datetime.date(y, m, d) <= datetime.date.today()
|
||
]
|
||
|
||
AK_INTERVAL = 0.8 # 每次调用后间隔秒 (防封 IP)
|
||
AK_TIMEOUT = 30.0 # 单次调用硬超时秒 (akshare 经常挂死)
|
||
AK_MAX_RETRIES = 3 # 单次下载重试次数 (含指数退避)
|
||
RETRY_BACKOFF = [2, 4, 8] # 重试间隔秒 (指数退避)
|
||
PROGRESS_LOG_EVERY = 100 # 每 N 个单位打一次进度日志
|
||
CIRCUIT_BREAKER = 30 # 连续失败 N 个 → 断路退出
|
||
|
||
# 模式 A: per-stock 类型 (8 类)
|
||
PER_STOCK_TYPES = (
|
||
"valuation", # stock_value_em(symbol="600519")
|
||
"northbound", # stock_hsgt_individual_em(symbol="600519")
|
||
"share_capital", # stock_share_change_cninfo(symbol="600519")
|
||
"balance", # stock_balance_sheet_by_report_em(symbol="SH600519")
|
||
"income", # stock_profit_sheet_by_report_em(symbol="SH600519")
|
||
"cashflow", # stock_cash_flow_sheet_by_report_em(symbol="SH600519")
|
||
"financial_abstract", # stock_financial_abstract(symbol="600519")
|
||
# top_holders 单独 (per-stock × per-period)
|
||
)
|
||
# 模式 B: per-date 类型 (4 类, margin_szse 跳过)
|
||
PER_DATE_TYPES = (
|
||
"dragon_tiger", # stock_lhb_detail_em(start_date, end_date)
|
||
"block_trade", # stock_dzjy_mrmx(symbol="A股", start_date, end_date)
|
||
"margin_sse", # stock_margin_detail_sse(date)
|
||
"restricted", # stock_restricted_release_detail_em(start_date, end_date)
|
||
)
|
||
# 模式 C: per-period 类型 (2 类)
|
||
PER_PERIOD_TYPES = (
|
||
"forecast", # stock_yjyg_em(date=period)
|
||
"express", # stock_yjkb_em(date=period)
|
||
)
|
||
# 模式 D: one-shot 类型 (2 类)
|
||
ONE_SHOT_TYPES = (
|
||
"index_const", # index_stock_cons_csindex(symbol) × 3 合并
|
||
"industry", # sw_index_first_info()
|
||
)
|
||
# top_holders 特殊: per-stock × per-period
|
||
TOP_HOLDERS = "top_holders"
|
||
|
||
ALL_TYPES = (
|
||
PER_STOCK_TYPES
|
||
+ (TOP_HOLDERS,)
|
||
+ PER_DATE_TYPES
|
||
+ PER_PERIOD_TYPES
|
||
+ ONE_SHOT_TYPES
|
||
)
|
||
|
||
|
||
# ======================== 日志 ========================
|
||
|
||
def setup_logging() -> Tuple[logging.Logger, Path]:
|
||
"""配置 root logger: 同时写 stdout + 文件。返回 (logger, log_file)。"""
|
||
LOG_DIR.mkdir(parents=True, exist_ok=True)
|
||
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
log_file = LOG_DIR / f"akshare_static_{ts}.log"
|
||
|
||
fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
|
||
root = logging.getLogger()
|
||
root.setLevel(logging.INFO)
|
||
for h in list(root.handlers):
|
||
root.removeHandler(h)
|
||
|
||
sh = logging.StreamHandler(sys.stdout)
|
||
sh.setFormatter(fmt)
|
||
root.addHandler(sh)
|
||
|
||
fh = logging.FileHandler(log_file, encoding="utf-8")
|
||
fh.setFormatter(fmt)
|
||
root.addHandler(fh)
|
||
|
||
return logging.getLogger(__name__), log_file
|
||
|
||
|
||
logger, LOG_FILE = setup_logging()
|
||
|
||
|
||
# ======================== 工具函数 ========================
|
||
|
||
def normalize_date(s: str) -> str:
|
||
"""接受 YYYYMMDD 或 YYYY-MM-DD, 返回 YYYYMMDD; 空串 → 今天。"""
|
||
s = (s or "").strip()
|
||
if not s:
|
||
return datetime.date.today().strftime("%Y%m%d")
|
||
s = s.replace("-", "")
|
||
if len(s) == 8 and s.isdigit():
|
||
return s
|
||
raise ValueError(f"无效日期格式: {s} (期望 YYYYMMDD 或 YYYY-MM-DD)")
|
||
|
||
|
||
def guess_exchange_by_code(code: str) -> str:
|
||
"""6/68/51/9 开头 → SH (含 9 开头 B 股), 其他 → SZ。
|
||
11/12 开头是可转债, 不应进入 (列表源 stock_info_a_code_name 只返股票)。"""
|
||
if code.startswith(("6", "68", "51", "9")):
|
||
return "SH"
|
||
return "SZ"
|
||
|
||
|
||
def code_to_symbol(code: str, exchange: str, endpoint: str) -> str:
|
||
"""同 code 在不同 akshare 端点格式不同。
|
||
|
||
- stock_value_em / northbound / financial_abstract / share_change_cninfo:
|
||
纯数字 "600519"
|
||
- balance/income/cashflow (三大报表): "SH600519" / "SZ000001" (大写前缀)
|
||
- top_holders (stock_gdfx_free_top_10_em): "sh600519" (小写前缀)
|
||
|
||
endpoint 取端点函数名 (作为标识符), 不区分大小写比较。
|
||
"""
|
||
e = endpoint.lower()
|
||
if e in ("balance_sheet", "income_sheet", "cashflow_sheet"):
|
||
# 三大报表: SH/SZ 大写前缀
|
||
return f"{exchange}{code}"
|
||
if e == "top_holders":
|
||
# 十大股东: 小写前缀
|
||
pfx = "sh" if exchange == "SH" else "sz"
|
||
return f"{pfx}{code}"
|
||
# 默认 (valuation / northbound / financial_abstract / share_capital): 纯数字
|
||
return code
|
||
|
||
|
||
# ======================== 超时 + 重试 ========================
|
||
|
||
def call_ak_with_timeout(
|
||
fn: Callable[..., pd.DataFrame],
|
||
*args: Any,
|
||
timeout: float = AK_TIMEOUT,
|
||
**kwargs: Any,
|
||
) -> pd.DataFrame:
|
||
"""单线程 ThreadPoolExecutor 包装 akshare 调用 + 硬超时。
|
||
|
||
akshare 内部 requests 在网络异常时会重试很久 (无总超时), 必须用
|
||
future-level timeout 才能保证不挂死。超时抛 concurrent.futures.TimeoutError
|
||
(上层捕获并计 failed)。
|
||
|
||
注意: 超时后 future 不可真正 kill (Python 线程不能强杀), 但 executor
|
||
退出后工作线程仍会在后台跑完或最终超时; 对进程主流程无影响 (我们不
|
||
等 join)。每次新建 executor 避免线程累积。
|
||
"""
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
|
||
future = ex.submit(fn, *args, **kwargs)
|
||
return future.result(timeout=timeout)
|
||
|
||
|
||
def call_ak_with_retry(
|
||
fn: Callable[..., pd.DataFrame],
|
||
unit_label: str,
|
||
*args: Any,
|
||
**kwargs: Any,
|
||
) -> Tuple[Optional[pd.DataFrame], str]:
|
||
"""带重试退避的 akshare 调用。返 (df_or_None, status)。
|
||
|
||
status ∈ {'ok', 'failed'}:
|
||
- 调用成功 (df 可能空) → ('ok' or 'empty', df)
|
||
- 重试耗尽仍异常/超时 → ('failed', None)
|
||
|
||
empty vs ok 的区分在调用方按 df.empty 判 (此处统一返 df, 调用方判 empty)。
|
||
重试退避: 第1次失败 sleep 2s, 第2次 4s, 第3次 8s (RETRY_BACKOFF)。
|
||
"""
|
||
last_err: Optional[Exception] = None
|
||
for attempt in range(AK_MAX_RETRIES):
|
||
try:
|
||
df = call_ak_with_timeout(fn, *args, **kwargs)
|
||
return df, "ok"
|
||
except concurrent.futures.TimeoutError as e:
|
||
last_err = e
|
||
logger.warning(
|
||
"%s 超时重试 %d/%d: (>%ds)",
|
||
unit_label, attempt + 1, AK_MAX_RETRIES, int(AK_TIMEOUT),
|
||
)
|
||
except Exception as e:
|
||
last_err = e
|
||
msg = str(e)[:200]
|
||
logger.warning(
|
||
"%s 异常重试 %d/%d: %s",
|
||
unit_label, attempt + 1, AK_MAX_RETRIES, msg,
|
||
)
|
||
# 指数退避
|
||
if attempt < AK_MAX_RETRIES - 1:
|
||
time.sleep(RETRY_BACKOFF[attempt])
|
||
logger.error("%s 重试 %d 次仍失败: %s", unit_label, AK_MAX_RETRIES, last_err)
|
||
return None, "failed"
|
||
|
||
|
||
# ======================== 股票列表 (akshare 轻量端点) ========================
|
||
|
||
def fetch_all_stocks() -> List[Tuple[str, str]]:
|
||
"""akshare stock_info_a_code_name() 拉全市场 A 股代码列表。
|
||
|
||
返回 [(code, exchange), ...], 例 [('600519', 'SH'), ('000001', 'SZ')]。
|
||
端点轻量, 单次调用返 ~5500 行 (code + name 两列), 不取行情。
|
||
exchange 按 code 开头猜 (6/68/51/9 → SH, 其他 → SZ)。
|
||
"""
|
||
df, status = call_ak_with_retry(
|
||
ak.stock_info_a_code_name, "all_stocks",
|
||
)
|
||
if status == "failed" or df is None:
|
||
raise RuntimeError("stock_info_a_code_name 拉取失败")
|
||
if df.empty:
|
||
raise RuntimeError("stock_info_a_code_name 返空 (异常)")
|
||
|
||
# akshare 返回列名: code, name (stock_info_a_code_name)
|
||
out: List[Tuple[str, str]] = []
|
||
n_skip = 0
|
||
for code in df["code"].tolist():
|
||
code = str(code).strip()
|
||
if len(code) != 6 or not code.isdigit():
|
||
n_skip += 1
|
||
continue
|
||
exchange = guess_exchange_by_code(code)
|
||
out.append((code, exchange))
|
||
logger.info(
|
||
"股票列表: 总 %d 只 (跳过 %d 非法 code), 来自 stock_info_a_code_name",
|
||
len(out), n_skip,
|
||
)
|
||
return out
|
||
|
||
|
||
# ======================== 路径 / marker ========================
|
||
|
||
def subdir_for(data_type: str) -> Path:
|
||
"""data_type → OUT_DIR / <subdir>。"""
|
||
return OUT_DIR / data_type
|
||
|
||
|
||
def parquet_path_per_unit(data_type: str, unit_id: str) -> Path:
|
||
"""通用 parquet 路径: <data_type>/<unit_id>.parquet。
|
||
|
||
unit_id 由各模式自拼:
|
||
- per-stock: '600519.SH_valuation'
|
||
- per-stock × per-period (top_holders): '600519.SH_2020930_top_holders'
|
||
- per-date: '20260715_dragon_tiger'
|
||
- per-period: '20251231_forecast'
|
||
- one-shot: 'index_const'
|
||
"""
|
||
return subdir_for(data_type) / f"{unit_id}.parquet"
|
||
|
||
|
||
def marker_path_for(parquet_path: Path) -> Path:
|
||
"""parquet → 同目录 .{stem}.akshare marker。"""
|
||
return parquet_path.parent / f".{parquet_path.stem}.akshare"
|
||
|
||
|
||
def load_done_units(data_type: str) -> set:
|
||
"""扫子目录 marker 构造已完成 unit_id 集合 (真相源)。
|
||
|
||
marker 文件名格式: .{unit_id}.akshare
|
||
直接用 stem (= unit_id) 作为 key, 不解析 (各模式 unit_id 规则不同, 统一字符串匹配)。
|
||
"""
|
||
done: set = set()
|
||
d = subdir_for(data_type)
|
||
if not d.exists():
|
||
return done
|
||
ext = ".akshare"
|
||
for marker in d.glob(f".*{ext}"):
|
||
name = marker.name
|
||
if not name.startswith(".") or not name.endswith(ext):
|
||
continue
|
||
unit_id = name[1:-len(ext)]
|
||
if unit_id:
|
||
done.add(unit_id)
|
||
return done
|
||
|
||
|
||
def write_parquet_and_marker(
|
||
df: pd.DataFrame,
|
||
parquet_path: Path,
|
||
) -> bool:
|
||
"""写 parquet + marker。返 True 成功 / False 失败。"""
|
||
try:
|
||
parquet_path.parent.mkdir(parents=True, exist_ok=True)
|
||
df.to_parquet(parquet_path, index=False)
|
||
marker_path_for(parquet_path).write_text(
|
||
datetime.datetime.now().isoformat()
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
logger.error("写入 %s 失败: %s", parquet_path, e)
|
||
return False
|
||
|
||
|
||
def _df_or_empty(result: Tuple[Optional[pd.DataFrame], str]) -> pd.DataFrame:
|
||
"""把 call_ak_with_retry 的返回 (df_or_None, status) 转成非 None df。
|
||
|
||
必须 None 显式判断 (不能 `df or pd.DataFrame()`, DataFrame 的 truth value
|
||
ambiguous, 会抛 "The truth value of a DataFrame is ambiguous"。
|
||
"""
|
||
df = result[0]
|
||
return df if df is not None else pd.DataFrame()
|
||
|
||
|
||
# ======================== per-stock fetch 函数 (8 类) ========================
|
||
|
||
def fetch_valuation(symbol: str) -> pd.DataFrame:
|
||
"""stock_value_em(symbol='600519') — 估值 (PE/PB/市值等13列, ~2000行/股)。"""
|
||
return _df_or_empty(call_ak_with_retry(
|
||
ak.stock_value_em, f"valuation/{symbol}", symbol=symbol,
|
||
))
|
||
|
||
|
||
def fetch_northbound(symbol: str) -> pd.DataFrame:
|
||
"""stock_hsgt_individual_em(symbol='600519') — 北向持股 (~1700行/股)。"""
|
||
return _df_or_empty(call_ak_with_retry(
|
||
ak.stock_hsgt_individual_em, f"northbound/{symbol}", symbol=symbol,
|
||
))
|
||
|
||
|
||
def fetch_share_capital(symbol: str) -> pd.DataFrame:
|
||
"""stock_share_change_cninfo(symbol='600519') — 股本变动 (44列)。"""
|
||
return _df_or_empty(call_ak_with_retry(
|
||
ak.stock_share_change_cninfo, f"share_capital/{symbol}", symbol=symbol,
|
||
))
|
||
|
||
|
||
def fetch_balance_sheet(symbol: str) -> pd.DataFrame:
|
||
"""stock_balance_sheet_by_report_em(symbol='SH600519') — 资产负债表 (319列)。"""
|
||
return _df_or_empty(call_ak_with_retry(
|
||
ak.stock_balance_sheet_by_report_em, f"balance/{symbol}", symbol=symbol,
|
||
))
|
||
|
||
|
||
def fetch_income_sheet(symbol: str) -> pd.DataFrame:
|
||
"""stock_profit_sheet_by_report_em(symbol='SH600519') — 利润表 (203列)。"""
|
||
return _df_or_empty(call_ak_with_retry(
|
||
ak.stock_profit_sheet_by_report_em, f"income/{symbol}", symbol=symbol,
|
||
))
|
||
|
||
|
||
def fetch_cashflow_sheet(symbol: str) -> pd.DataFrame:
|
||
"""stock_cash_flow_sheet_by_report_em(symbol='SH600519') — 现金流量表 (254列)。"""
|
||
return _df_or_empty(call_ak_with_retry(
|
||
ak.stock_cash_flow_sheet_by_report_em, f"cashflow/{symbol}", symbol=symbol,
|
||
))
|
||
|
||
|
||
def fetch_financial_abstract(symbol: str) -> pd.DataFrame:
|
||
"""stock_financial_abstract(symbol='600519') — 财务摘要 (80指标)。"""
|
||
return _df_or_empty(call_ak_with_retry(
|
||
ak.stock_financial_abstract, f"financial_abstract/{symbol}", symbol=symbol,
|
||
))
|
||
|
||
|
||
# ======================== top_holders (per-stock × per-period) ========================
|
||
|
||
def fetch_top_holders_one_period(
|
||
symbol: str, period: str,
|
||
) -> pd.DataFrame:
|
||
"""stock_gdfx_free_top_10_em(symbol='sh600519', date='20250930') 单期。
|
||
symbol 小写前缀, period YYYYMMDD."""
|
||
return _df_or_empty(call_ak_with_retry(
|
||
ak.stock_gdfx_free_top_10_em,
|
||
f"top_holders/{symbol}/{period}",
|
||
symbol=symbol, date=period,
|
||
))
|
||
|
||
|
||
# ======================== per-date fetch 函数 (4 类) ========================
|
||
|
||
def fetch_dragon_tiger(date: str) -> pd.DataFrame:
|
||
"""stock_lhb_detail_em(start_date=end_date=date) — 单日龙虎榜。"""
|
||
return _df_or_empty(call_ak_with_retry(
|
||
ak.stock_lhb_detail_em, f"dragon_tiger/{date}",
|
||
start_date=date, end_date=date,
|
||
))
|
||
|
||
|
||
def fetch_block_trade(date: str) -> pd.DataFrame:
|
||
"""stock_dzjy_mrmx(symbol='A股', start_date=end_date=date) — 大宗交易明细。"""
|
||
return _df_or_empty(call_ak_with_retry(
|
||
ak.stock_dzjy_mrmx, f"block_trade/{date}",
|
||
symbol="A股", start_date=date, end_date=date,
|
||
))
|
||
|
||
|
||
def fetch_margin_sse(date: str) -> pd.DataFrame:
|
||
"""stock_margin_detail_sse(date=date) — 沪市融资融券明细。"""
|
||
return _df_or_empty(call_ak_with_retry(
|
||
ak.stock_margin_detail_sse, f"margin_sse/{date}", date=date,
|
||
))
|
||
|
||
|
||
def fetch_restricted(date: str) -> pd.DataFrame:
|
||
"""stock_restricted_release_detail_em(start_date=end_date=date) — 解禁明细。"""
|
||
return _df_or_empty(call_ak_with_retry(
|
||
ak.stock_restricted_release_detail_em, f"restricted/{date}",
|
||
start_date=date, end_date=date,
|
||
))
|
||
|
||
|
||
# ======================== per-period fetch 函数 (2 类) ========================
|
||
|
||
def fetch_forecast(period: str) -> pd.DataFrame:
|
||
"""stock_yjyg_em(date=period) — 业绩预告 (全市场, 单期单调用)。"""
|
||
return _df_or_empty(call_ak_with_retry(
|
||
ak.stock_yjyg_em, f"forecast/{period}", date=period,
|
||
))
|
||
|
||
|
||
def fetch_express(period: str) -> pd.DataFrame:
|
||
"""stock_yjkb_em(date=period) — 业绩快报 (全市场, 单期单调用)。"""
|
||
return _df_or_empty(call_ak_with_retry(
|
||
ak.stock_yjkb_em, f"express/{period}", date=period,
|
||
))
|
||
|
||
|
||
# ======================== one-shot fetch 函数 (2 类) ========================
|
||
|
||
def fetch_index_const() -> pd.DataFrame:
|
||
"""index_stock_cons_csindex 循环 ["000300","000905","000852"] 合并。
|
||
|
||
300/500/1000 三大指数成分, 合并返单 df, 加 index_code 列标识来源。
|
||
"""
|
||
index_codes = ["000300", "000905", "000852"]
|
||
frames: List[pd.DataFrame] = []
|
||
for idx in index_codes:
|
||
df, status = call_ak_with_retry(
|
||
ak.index_stock_cons_csindex, f"index_const/{idx}", symbol=idx,
|
||
)
|
||
if status == "failed" or df is None:
|
||
logger.warning("index_const %s 失败, 跳过该指数", idx)
|
||
continue
|
||
if df.empty:
|
||
logger.warning("index_const %s 返空, 跳过", idx)
|
||
continue
|
||
df = df.copy()
|
||
df["index_code"] = idx
|
||
frames.append(df)
|
||
time.sleep(AK_INTERVAL) # 三次调用之间也限速
|
||
if not frames:
|
||
return pd.DataFrame()
|
||
return pd.concat(frames, ignore_index=True)
|
||
|
||
|
||
def fetch_industry() -> pd.DataFrame:
|
||
"""sw_index_first_info() — 申万一级行业列表 (东财接口 ConnectionError, 申万替代)。"""
|
||
return _df_or_empty(call_ak_with_retry(
|
||
ak.sw_index_first_info, "industry",
|
||
))
|
||
|
||
|
||
# ======================== 通用下载单元 (写 parquet + marker) ========================
|
||
|
||
def download_one_unit(
|
||
data_type: str,
|
||
unit_id: str,
|
||
fetch_fn: Callable[[], pd.DataFrame],
|
||
force: bool,
|
||
) -> Tuple[str, int]:
|
||
"""通用单 unit 下载: 拉 df → 写 parquet + marker。
|
||
|
||
fetch_fn() → df (可能空) 或 raise (call_ak_with_retry 已吞异常返 None,
|
||
各 fetch_xxx 已把 None 转空 df; 这里 df 永远非 None 但可能空)。
|
||
|
||
返 (status, rows), status ∈ {'ok', 'skipped', 'empty', 'failed'}。
|
||
"""
|
||
parquet_path = parquet_path_per_unit(data_type, unit_id)
|
||
marker_path = marker_path_for(parquet_path)
|
||
|
||
if not force and marker_path.exists():
|
||
return "skipped", 0
|
||
|
||
try:
|
||
df = fetch_fn()
|
||
except Exception as e:
|
||
# 兜底: call_ak_with_retry 内部已重试, 这里理论上不应触发,
|
||
# 但保险 (各 fetch_xxx 转 None→空 df, 这里捕获意外异常)
|
||
logger.error("[%s] %s fetch 异常: %s", data_type, unit_id, e)
|
||
return "failed", 0
|
||
|
||
if df is None:
|
||
# fetch_xxx 保证返非 None, 但保险
|
||
return "failed", 0
|
||
|
||
# 写 parquet + marker (空 df 也写, 静态语义: "查过了确实无数据")
|
||
if not write_parquet_and_marker(df, parquet_path):
|
||
return "failed", 0
|
||
return ("ok" if not df.empty else "empty"), len(df)
|
||
|
||
|
||
# ======================== 主循环 (通用, 适用所有四种模式) ========================
|
||
|
||
def run_one_type(
|
||
data_type: str,
|
||
units: List[Tuple[str, Callable[[], pd.DataFrame]]],
|
||
args: argparse.Namespace,
|
||
) -> Tuple[dict, bool]:
|
||
"""通用类型主循环: 按顺序处理每个 unit (unit_id, fetch_fn)。
|
||
|
||
返 (stats, circuit_triggered)。
|
||
|
||
每个 unit 之间 sleep AK_INTERVAL 限速 (最后一个不 sleep)。
|
||
"""
|
||
# marker 断点续传
|
||
done_set = load_done_units(data_type)
|
||
if args.force:
|
||
todo = [(uid, fn) for uid, fn in units]
|
||
else:
|
||
todo = [(uid, fn) for uid, fn in units if uid not in done_set]
|
||
logger.info(
|
||
"[%s] 待处理 %d (已完成 %d, 总 %d)",
|
||
data_type, len(todo), len(done_set), len(units),
|
||
)
|
||
|
||
stats = {"ok": 0, "skipped": 0, "empty": 0, "failed": 0, "rows": 0}
|
||
consec_fail = 0
|
||
circuit_triggered = False
|
||
t_start = time.time()
|
||
total_bytes = 0
|
||
|
||
for i, (uid, fn) in enumerate(todo):
|
||
try:
|
||
status, rows = download_one_unit(data_type, uid, fn, args.force)
|
||
except Exception as e:
|
||
status, rows = "failed", 0
|
||
logger.debug("[%s] %s 异常: %s", data_type, uid, e)
|
||
|
||
stats[status] = stats.get(status, 0) + 1
|
||
if status == "ok":
|
||
stats["rows"] += rows
|
||
consec_fail = 0
|
||
# 统计 parquet 大小
|
||
try:
|
||
total_bytes += parquet_path_per_unit(data_type, uid).stat().st_size
|
||
except OSError:
|
||
pass
|
||
elif status == "empty":
|
||
consec_fail = 0
|
||
elif status == "failed":
|
||
consec_fail += 1
|
||
# skipped 不重置也不递增
|
||
|
||
# 断路器
|
||
if consec_fail >= CIRCUIT_BREAKER:
|
||
logger.error(
|
||
"[%s] [FATAL] 断路器触发: 连续 %d 个 unit 失败, "
|
||
"akshare 疑似不可达, 退出 (done_set 不含 failed, 复跑会重试)",
|
||
data_type, consec_fail,
|
||
)
|
||
circuit_triggered = True
|
||
break
|
||
|
||
# 进度日志
|
||
if (i + 1) % PROGRESS_LOG_EVERY == 0 or (i + 1) == len(todo):
|
||
elapsed = time.time() - t_start
|
||
logger.info(
|
||
"[%s] 进度 %d/%d ok=%d empty=%d failed=%d skipped=%d "
|
||
"rows=%d size=%.1fMB (%.0f秒)",
|
||
data_type, i + 1, len(todo),
|
||
stats["ok"], stats["empty"], stats["failed"], stats["skipped"],
|
||
stats["rows"], total_bytes / 1024 / 1024, elapsed,
|
||
)
|
||
|
||
# 限速
|
||
if i < len(todo) - 1:
|
||
time.sleep(AK_INTERVAL)
|
||
|
||
elapsed = time.time() - t_start
|
||
logger.info(
|
||
"[%s] 完成, 耗时 %.1f 秒 (%.1f分), 统计: %s, 总 parquet 大小 %.1fMB",
|
||
data_type, elapsed, elapsed / 60,
|
||
json.dumps(stats, ensure_ascii=False),
|
||
total_bytes / 1024 / 1024,
|
||
)
|
||
return stats, circuit_triggered
|
||
|
||
|
||
# ======================== 各模式 units 构造 ========================
|
||
|
||
def build_per_stock_units(
|
||
data_type: str,
|
||
endpoint: str,
|
||
fetch_binder: Callable[[str], pd.DataFrame],
|
||
all_codes: List[Tuple[str, str]],
|
||
args: argparse.Namespace,
|
||
) -> List[Tuple[str, Callable[[], pd.DataFrame]]]:
|
||
"""构造 per-stock units: [(unit_id, fetch_fn), ...]。
|
||
|
||
unit_id = '{code}.{EXC}_{data_type}', 例 '600519.SH_valuation'
|
||
fetch_fn 闭包绑定 symbol (按 endpoint 格式)。
|
||
|
||
endpoint 决定 symbol 格式 (见 code_to_symbol)。
|
||
"""
|
||
# --codes 过滤
|
||
if args.codes:
|
||
codes_set = {c.strip() for c in args.codes.split(",") if c.strip()}
|
||
todo_codes = [(c, guess_exchange_by_code(c)) for c in codes_set]
|
||
else:
|
||
todo_codes = list(all_codes)
|
||
# --limit 截断
|
||
if args.limit > 0:
|
||
todo_codes = todo_codes[: args.limit]
|
||
|
||
units: List[Tuple[str, Callable[[], pd.DataFrame]]] = []
|
||
for code, exc in todo_codes:
|
||
symbol = code_to_symbol(code, exc, endpoint)
|
||
unit_id = f"{code}.{exc}_{data_type}"
|
||
# partial 绑定 symbol (lambda 闭包易 late-bind, 用 partial 安全)
|
||
fn = partial(fetch_binder, symbol)
|
||
units.append((unit_id, fn))
|
||
return units
|
||
|
||
|
||
def build_top_holders_units(
|
||
all_codes: List[Tuple[str, str]],
|
||
args: argparse.Namespace,
|
||
) -> List[Tuple[str, Callable[[], pd.DataFrame]]]:
|
||
"""top_holders 特殊: per-stock × per-period (近5年×4季 = 20期).
|
||
|
||
unit_id = '{code}.{EXC}_{period}_top_holders', 例 '600519.SH_2020930_top_holders'
|
||
"""
|
||
if args.codes:
|
||
codes_set = {c.strip() for c in args.codes.split(",") if c.strip()}
|
||
todo_codes = [(c, guess_exchange_by_code(c)) for c in codes_set]
|
||
else:
|
||
todo_codes = list(all_codes)
|
||
if args.limit > 0:
|
||
todo_codes = todo_codes[: args.limit]
|
||
|
||
# 报告期取近 5 年×4季 (REPORT_PERIODS 已含至今所有)
|
||
periods = REPORT_PERIODS[-20:] if len(REPORT_PERIODS) >= 20 else REPORT_PERIODS
|
||
|
||
units: List[Tuple[str, Callable[[], pd.DataFrame]]] = []
|
||
for code, exc in todo_codes:
|
||
symbol = code_to_symbol(code, exc, "top_holders")
|
||
for period in periods:
|
||
unit_id = f"{code}.{exc}_{period}_{TOP_HOLDERS}"
|
||
fn = partial(fetch_top_holders_one_period, symbol, period)
|
||
units.append((unit_id, fn))
|
||
logger.info(
|
||
"[top_holders] %d 票 × %d 期 = %d units",
|
||
len(todo_codes), len(periods), len(units),
|
||
)
|
||
return units
|
||
|
||
|
||
def build_per_date_units(
|
||
data_type: str,
|
||
fetch_fn: Callable[[str], pd.DataFrame],
|
||
args: argparse.Namespace,
|
||
) -> List[Tuple[str, Callable[[], pd.DataFrame]]]:
|
||
"""构造 per-date units: 每个交易日 1 个 unit。
|
||
|
||
unit_id = '{date}_{data_type}', 例 '20260715_dragon_tiger'
|
||
|
||
交易日简单生成 (周一到周五), 排除节假日策略: 拉到空就 skip (计 empty,
|
||
不计 failed); 不依赖节假日表 (节假日多日无数据, 拉空即合法)。
|
||
"""
|
||
start = normalize_date(args.start)
|
||
end = normalize_date(args.end)
|
||
start_dt = datetime.datetime.strptime(start, "%Y%m%d").date()
|
||
end_dt = datetime.datetime.strptime(end, "%Y%m%d").date()
|
||
if start_dt > end_dt:
|
||
raise ValueError(f"--start {start} > --end {end}")
|
||
|
||
units: List[Tuple[str, Callable[[], pd.DataFrame]]] = []
|
||
d = start_dt
|
||
while d <= end_dt:
|
||
# 周一到周五 (周末无交易, 不入列)
|
||
if d.weekday() < 5:
|
||
date_str = d.strftime("%Y%m%d")
|
||
unit_id = f"{date_str}_{data_type}"
|
||
units.append((unit_id, partial(fetch_fn, date_str)))
|
||
d += datetime.timedelta(days=1)
|
||
logger.info(
|
||
"[%s] 日期范围 %s ~ %s, 工作日 %d 天 (节假日拉空计 empty)",
|
||
data_type, start, end, len(units),
|
||
)
|
||
return units
|
||
|
||
|
||
def build_per_period_units(
|
||
data_type: str,
|
||
fetch_fn: Callable[[str], pd.DataFrame],
|
||
) -> List[Tuple[str, Callable[[], pd.DataFrame]]]:
|
||
"""构造 per-period units: 每个报告期 1 个 unit。
|
||
|
||
unit_id = '{period}_{data_type}', 例 '20251231_forecast'
|
||
报告期取 REPORT_PERIODS (近 5 年×4季)。
|
||
"""
|
||
units: List[Tuple[str, Callable[[], pd.DataFrame]]] = []
|
||
for period in REPORT_PERIODS:
|
||
unit_id = f"{period}_{data_type}"
|
||
units.append((unit_id, partial(fetch_fn, period)))
|
||
logger.info(
|
||
"[%s] 报告期 %d 个 (近5年×4季, %s..%s)",
|
||
data_type, len(units), REPORT_PERIODS[0], REPORT_PERIODS[-1],
|
||
)
|
||
return units
|
||
|
||
|
||
def build_one_shot_units(
|
||
data_type: str,
|
||
fetch_fn: Callable[[], pd.DataFrame],
|
||
) -> List[Tuple[str, Callable[[], pd.DataFrame]]]:
|
||
"""构造 one-shot units: 单个 unit。
|
||
|
||
unit_id = data_type, 例 'index_const'
|
||
"""
|
||
return [(data_type, fetch_fn)]
|
||
|
||
|
||
# ======================== CLI / main ========================
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
p = argparse.ArgumentParser(
|
||
description="AKShare A 股静态数据全量下载 (16 类, 四种模式)",
|
||
)
|
||
p.add_argument(
|
||
"--types", default=",".join(ALL_TYPES),
|
||
help=f"逗号分隔类型, 默认全部 ({','.join(ALL_TYPES)})",
|
||
)
|
||
p.add_argument(
|
||
"--start", default=START_DATE_DEFAULT,
|
||
help=f"起始日期 YYYYMMDD 或 YYYY-MM-DD, 默认 {START_DATE_DEFAULT}",
|
||
)
|
||
p.add_argument("--end", default="", help="结束日期, 默认今天")
|
||
p.add_argument(
|
||
"--codes",
|
||
help="指定代码逗号分隔 (6 位无前缀), 如 600519,000001 (仅 per-stock 类生效)",
|
||
)
|
||
p.add_argument(
|
||
"--limit", type=int, default=0,
|
||
help="限制处理股票数 (per-stock 类生效), 测试用",
|
||
)
|
||
p.add_argument("--force", action="store_true", help="强制重下, 忽略 marker")
|
||
return p.parse_args()
|
||
|
||
|
||
def resolve_types(s: str) -> List[str]:
|
||
"""解析 --types 字符串 → 去重保序的类型列表。无效类型 SystemExit。"""
|
||
parts = [t.strip() for t in s.split(",") if t.strip()]
|
||
bad = [t for t in parts if t not in ALL_TYPES]
|
||
if bad:
|
||
raise SystemExit(f"未知 --types: {bad}, 可选 {list(ALL_TYPES)}")
|
||
if not parts:
|
||
return list(ALL_TYPES)
|
||
seen: set = set()
|
||
out: List[str] = []
|
||
for t in parts:
|
||
if t not in seen:
|
||
out.append(t)
|
||
seen.add(t)
|
||
return out
|
||
|
||
|
||
def run_type_dispatch(
|
||
t: str,
|
||
all_codes: List[Tuple[str, str]],
|
||
args: argparse.Namespace,
|
||
) -> Tuple[dict, bool]:
|
||
"""按类型路由到对应模式 + fetch 函数。返 (stats, circuit_triggered)。"""
|
||
if t in PER_STOCK_TYPES:
|
||
endpoint_map = {
|
||
"valuation": ("valuation", fetch_valuation),
|
||
"northbound": ("northbound", fetch_northbound),
|
||
"share_capital": ("share_capital", fetch_share_capital),
|
||
"balance": ("balance_sheet", fetch_balance_sheet),
|
||
"income": ("income_sheet", fetch_income_sheet),
|
||
"cashflow": ("cashflow_sheet", fetch_cashflow_sheet),
|
||
"financial_abstract": ("financial_abstract", fetch_financial_abstract),
|
||
}
|
||
endpoint, fetch_fn = endpoint_map[t]
|
||
units = build_per_stock_units(t, endpoint, fetch_fn, all_codes, args)
|
||
return run_one_type(t, units, args)
|
||
|
||
if t == TOP_HOLDERS:
|
||
units = build_top_holders_units(all_codes, args)
|
||
return run_one_type(t, units, args)
|
||
|
||
if t in PER_DATE_TYPES:
|
||
fetch_map = {
|
||
"dragon_tiger": fetch_dragon_tiger,
|
||
"block_trade": fetch_block_trade,
|
||
"margin_sse": fetch_margin_sse,
|
||
"restricted": fetch_restricted,
|
||
}
|
||
units = build_per_date_units(t, fetch_map[t], args)
|
||
return run_one_type(t, units, args)
|
||
|
||
if t in PER_PERIOD_TYPES:
|
||
fetch_map = {
|
||
"forecast": fetch_forecast,
|
||
"express": fetch_express,
|
||
}
|
||
units = build_per_period_units(t, fetch_map[t])
|
||
return run_one_type(t, units, args)
|
||
|
||
if t in ONE_SHOT_TYPES:
|
||
fetch_map = {
|
||
"index_const": fetch_index_const,
|
||
"industry": fetch_industry,
|
||
}
|
||
units = build_one_shot_units(t, fetch_map[t])
|
||
return run_one_type(t, units, args)
|
||
|
||
logger.error("未知类型 (跳过): %s", t)
|
||
return {"ok": 0, "skipped": 0, "empty": 0, "failed": 0, "rows": 0}, False
|
||
|
||
|
||
def main() -> None:
|
||
args = parse_args()
|
||
start_date = normalize_date(args.start)
|
||
end_date = normalize_date(args.end)
|
||
types = resolve_types(args.types)
|
||
|
||
logger.info("=" * 60)
|
||
logger.info("AKShare A 股静态数据全量下载 (16 类, 四种模式)")
|
||
logger.info(" 输出目录: %s", OUT_DIR)
|
||
logger.info(" 日志文件: %s", LOG_FILE)
|
||
logger.info(" 日期范围: %s ~ %s", start_date, end_date)
|
||
logger.info(" 报告期数: %d (%s..%s)",
|
||
len(REPORT_PERIODS),
|
||
REPORT_PERIODS[0] if REPORT_PERIODS else "-",
|
||
REPORT_PERIODS[-1] if REPORT_PERIODS else "-")
|
||
logger.info(" 类型: %s", types)
|
||
logger.info(" 限速: 单线程串行, AK_INTERVAL=%.1fs, AK_TIMEOUT=%.0fs",
|
||
AK_INTERVAL, AK_TIMEOUT)
|
||
logger.info(" 当前时间: %s", datetime.datetime.now().isoformat())
|
||
logger.info("=" * 60)
|
||
|
||
# 检测 akshare 版本 (日志)
|
||
try:
|
||
logger.info("akshare 版本: %s", ak.__version__)
|
||
except AttributeError:
|
||
logger.info("akshare 版本: 未知 (无 __version__ 属性)")
|
||
|
||
any_circuit = False
|
||
# 是否需要股票列表
|
||
need_codes = any(
|
||
t in PER_STOCK_TYPES or t == TOP_HOLDERS for t in types
|
||
)
|
||
all_codes: List[Tuple[str, str]] = []
|
||
|
||
if need_codes:
|
||
try:
|
||
all_codes = fetch_all_stocks()
|
||
except Exception as e:
|
||
logger.error("[FATAL] 获取股票列表失败: %s", e)
|
||
sys.exit(1)
|
||
|
||
for t in types:
|
||
logger.info("-" * 50)
|
||
logger.info(">>> 类型: %s", t)
|
||
try:
|
||
stats, circuit = run_type_dispatch(t, all_codes, args)
|
||
except Exception as e:
|
||
logger.exception("[%s] 类型执行异常: %s", t, e)
|
||
any_circuit = True
|
||
break
|
||
if circuit:
|
||
any_circuit = True
|
||
logger.error("[%s] 断路器触发, 跳过后续类型", t)
|
||
break
|
||
|
||
logger.info("=" * 60)
|
||
logger.info("全部完成, 退出码 %d", 2 if any_circuit else 0)
|
||
logger.info("=" * 60)
|
||
sys.exit(2 if any_circuit else 0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|