From 66a2c9e9ce42c6125fbaa050420048ae1be75ab3 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 28 Jul 2026 21:54:46 +0800 Subject: [PATCH] =?UTF-8?q?feat(data):=20=E4=B8=AD=E8=AF=81=E6=8C=87?= =?UTF-8?q?=E6=95=B0=E6=97=A5=E7=BA=BF=E7=82=B9=E4=BD=8D=E4=BF=AE=E8=82=A1?= =?UTF-8?q?=E7=A5=A8/=E6=8C=87=E6=95=B0=E5=90=8C=E5=90=8D=E7=A2=B0?= =?UTF-8?q?=E6=92=9E=20+=20=E8=A1=8C=E4=B8=9A=E6=B2=BB=E5=81=8F=E5=B7=AE?= =?UTF-8?q?=20section=20=E5=90=8D=E6=A0=A1=E6=AD=A3/=E5=A2=9E=E9=87=8F?= =?UTF-8?q?=E6=8E=A5=E7=BA=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 行情(修碰撞): 新 sina_index_eod.py 拉 14 中证指数日线点位(000016/300/852/905/985/000928-937) 入 dbbardata exchange=SSE, 与 SZSE 同名股票分离(provider .XSHG→SSE 零改动命中)。sina 主源 + 腾讯兜底 5 个 sina 停 2016 的(000929/930/936/937/985); 东财两端点持续封 VPS。schtask sanguo-idx-eod 18:30 增量(busy_timeout 60s 防 bs_eod 撞锁)。策略03 行业 mavg30 不再读个股股价。 治偏差(行业): INDEX_SECTION_MAP section 名校正(中证800能源->中证能源等, 1208 公告 content 实证); index_monthly_wrapper STEP0b 接进 --indices 000928-937 月度增量(闭合 G1+G2 一次性未接增量缺口)。 注: 宽基治偏差(000852/000905/000016 等)早已完成(prior b237c2d); 行业指数 csindex 无 PDF 调整 公告(数据埋 HTML content), was_removed 仍~0, content 解析待专项。 --- .../data_platform/index_monthly_wrapper.ps1 | 8 + .../data_platform/parse_csindex_announce.py | 36 ++- scripts/data_platform/sina_index_eod.py | 289 ++++++++++++++++++ .../data_platform/sina_index_eod_wrapper.ps1 | 13 + 4 files changed, 334 insertions(+), 12 deletions(-) create mode 100644 scripts/data_platform/sina_index_eod.py create mode 100644 scripts/data_platform/sina_index_eod_wrapper.ps1 diff --git a/scripts/data_platform/index_monthly_wrapper.ps1 b/scripts/data_platform/index_monthly_wrapper.ps1 index 3be22b4..a5a1dbf 100644 --- a/scripts/data_platform/index_monthly_wrapper.ps1 +++ b/scripts/data_platform/index_monthly_wrapper.ps1 @@ -23,6 +23,14 @@ $dir = 'C:\sanguo_vnpy_v2\scripts\data_platform' & $py -X utf8 "$dir\parse_csindex_announce.py" --only both *>> $log "[$ts] STEP0 done exit=$LASTEXITCODE (失败不阻塞, migrate 回退 snapshot)" | Out-File $log -Append -Encoding UTF8 +# STEP0b: 中证一级行业指数系列 (000928-000937) 治偏差 - 独立第二次调用 +# 原因: main() 里 --indices 与 --only 互斥 (line ~683 早 return), 必须独立调用。 +# 不加 --max-notices-per-index: process_generic_index 每月从零重建 announce_union, 需保 was_removed 全历史。 +# 失败不阻塞 (旧 parquet 兜底, migrate 用现存 announce_union)。 +"[$ts] STEP0b parse_csindex_announce --indices 000928..000937 (中证行业指数系列, 独立调用)" | Out-File $log -Append -Encoding UTF8 +& $py -X utf8 "$dir\parse_csindex_announce.py" --indices 000928,000929,000930,000931,000932,000933,000934,000935,000936,000937 *>> $log +"[$ts] STEP0b done exit=$LASTEXITCODE (失败不阻塞, 旧 parquet 兜底)" | Out-File $log -Append -Encoding UTF8 + "[$ts] STEP1 index_const_hist_download --parts A B (深证 adjust + 中证 snapshot)" | Out-File $log -Append -Encoding UTF8 & $py -X utf8 "$dir\index_const_hist_download.py" --parts A B *>> $log "[$ts] STEP1 done exit=$LASTEXITCODE" | Out-File $log -Append -Encoding UTF8 diff --git a/scripts/data_platform/parse_csindex_announce.py b/scripts/data_platform/parse_csindex_announce.py index 9884c60..4220190 100644 --- a/scripts/data_platform/parse_csindex_announce.py +++ b/scripts/data_platform/parse_csindex_announce.py @@ -562,24 +562,36 @@ def parse_launch_xlsx(path: Path, index_code: str, publish_date: str, notice_id: # ======================== 通用指数处理 (G1+G2 扩展) ======================== # 内置指数 -> PDF section 名映射 (用于 parse_pdf_adjustments 精确定位 section) # 实证 csindex 公告系统 PDF section 用指数简称 (无"指数"后缀), 如 "中证1000 指数样本调整名单" -# 行业指数系列简称实证: akshare 返回 "800能源/800材料/..." (csindex PDF section 多用"中证800能源"全称) +# 行业指数系列 section 名实证 (2026-07-28): +# 来源: csindex 公告 id=1208 (2009-12-14 "中证行业指数调整名单") HTML content 嵌入文本 +# method: regex `(中证[一-龥]{1,10}指数)样本调整名单` 对 content 去标签后文本匹配 +# 实证 10 个 section 名: 中证能源/中证原材料/中证工业/中证可选消费/中证主要消费/ +# 中证医药卫生/中证金融地产/中证信息技术/中证电信/中证公用事业 +# 注意: akshare 当前 000928 简称="800能源" (后期改名中证800行业系列), +# 但 csindex 公告 content PDF section 文本仍用"中证能源"等早期名称。 +# parse_pdf_adjustments header_re = `中证[一-龥]{1,10}\s*指数样本调整名单` 贪婪回溯后 +# group(1) = "中证能源" 等 (剔"指数"后缀), 故 MAP 值不带"指数"二字。 +# ⚠️ 已知限制: csindex 对 000928-000937 几乎无 PDF 调整公告 (fetch_notices_by_index +# 只 6 条, theme=指数调样 3 条均无附件), 此 MAP 主要供未来潜在 PDF / HTML content +# 解析扩展使用。当前 was_removed 治偏差增益来自 xlsx 单指数临时调整 (不经此 MAP)。 INDEX_SECTION_MAP: Dict[str, str] = { # 已有 (CSI 1000/2000 走 main --only 路径, 不走 --indices) "000852": "中证1000", "932000": "中证2000", # G1: 中证全指 (全市场池, 解锁策略 02) "000985": "中证全指", - # G2: 中证一级行业指数 (行业轮动, 解锁策略 03) - 中证 800 行业指数系列 - "000928": "中证800能源", - "000929": "中证800材料", - "000930": "中证800工业", - "000931": "中证800可选", - "000932": "中证800消费", - "000933": "中证800医药", - "000934": "中证800金融", - "000935": "中证800信息", - "000936": "中证800通信", - "000937": "中证800公用", + # G2: 中证一级行业指数 (行业轮动, 解锁策略 03) - 中证行业指数系列 (000928-000937) + # section 名 1208 公告 HTML content 实证 (见上方注释) + "000928": "中证能源", + "000929": "中证原材料", + "000930": "中证工业", + "000931": "中证可选消费", + "000932": "中证主要消费", + "000933": "中证医药卫生", + "000934": "中证金融地产", + "000935": "中证信息技术", + "000936": "中证电信", + "000937": "中证公用事业", # 000938 不是行业指数 (中证民企ESG 50 等), 留作可选 } diff --git a/scripts/data_platform/sina_index_eod.py b/scripts/data_platform/sina_index_eod.py new file mode 100644 index 0000000..cda5078 --- /dev/null +++ b/scripts/data_platform/sina_index_eod.py @@ -0,0 +1,289 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""sina_index_eod.py — sanguo-idx-eod (18:30 schtask): 中证指数日线点位 -> dbbardata('d', SSE)。 + +背景: + dbbardata 6 位码同时是中证指数码与 SZSE 真实股票码 (000928~000937/000852/000905/ + 000016/000985), 历史只灌了 SZSE 股票行, 中证指数点位从未入库。000300 无碰撞, + 已有 SSE 指数行 (close 几千)。本脚本把 14 个中证指数点位灌入 exchange='SSE', + 与 SZSE 股票行 (exchange='SZSE') 隔离 — provider 的 .XSHG->SSE 映射本就如此。 + +数据源 (实证 2026-07-28, 级联取第一个"最近 30 天内有数据"的源): + 主 sina: ak.stock_zh_index_daily(symbol="sh000928") + cols=[date(str), open, high, low, close, volume] + 9/14 良好; 000929/000930/000936/000937/000985 sina 停在 2016-06-13 (源残缺)。 + 兜底1 东财: ak.index_zh_a_hist(symbol, period="daily") — VPS IP 持续 RemoteDisconnected。 + 兜底2 腾讯: ak.stock_zh_index_daily_tx(symbol="sh000929") — schema 不同! + cols=[date(date 对象), open, close, high, low, amount] (注意: 无 volume, amount 是成交额, + 列顺序 close 在 high 前)。对全 14 都有数据, 是当前兜底主力。 + 字段映射: volume 取 sina/东财 的 volume, 腾讯 volume=0; turnover=腾讯 amount, 其他源 0。 + 策略只用 close, volume/turnover 不影响; OHLC 三源都有。 + +源选择 (surgical): 不把腾讯改 primary 是为了保留 9 个 sina 良好 code 的 volume 数据 (腾讯无 volume 列, +切 primary 会把已入库的 volume 清零, 是回归)。腾讯仅在 sina stale 时兜底。 + +幂等: + dbbardata UNIQUE(symbol,exchange,datetime,interval) -> INSERT OR REPLACE 自动幂等。 + 只写 exchange='SSE' 行, 绝不动 exchange='SZSE' 股票行。 + +硬约束 (用户铁律): + 顶部清 http_proxy/https_proxy/all_proxy (直连不走代理); + 单线程, 每次 akshare 调用 sleep 1.2s; 禁止并发; 腾讯 tqdm 进度条不阻塞 (函数会跑完)。 + +用法: + python sina_index_eod.py [--lookback-days N] [--dry-run] # N 默认 30 (env LOOKBACK_DAYS) + python sina_index_eod.py --lookback-days 3650 # 首次回灌 +""" +import argparse +import os +import sys +import time + +# 硬约束: 直连不走代理 (akshare sina/eastmoney 都是国内源, 走代理反而挂) +for _k in ("http_proxy", "https_proxy", "all_proxy", + "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY"): + os.environ.pop(_k, None) + +import sqlite3 + +import pandas as pd + +# 同目录 import (与 xt_eod/bs_eod 一致, schtask 工作目录 = scripts/data_platform) +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from dbbardata_utils import normalize_daily_dt # noqa: E402 + +DB = r"C:\sanguo_vnpy_v2\data\quant_trading.db" +T0 = time.time() + +# 14 个中证指数 (与 SZSE 股票码碰撞, 仅灌 SSE 点位行) +CODES = [ + "000928", "000929", "000930", "000931", "000932", "000933", + "000934", "000935", "000936", "000937", + "000852", "000905", "000016", "000985", +] + +SINA_SLEEP = 1.2 # 单线程限速: 每次 akshare 调用后 sleep (避封 IP) +FALLBACK_RECENT_DAYS = 30 # sina 最后日期 < today-30 才触发东财兜底 + + +def log(m): + print(f"[IDX-EOD {time.time()-T0:.0f}s] {m}", flush=True) + + +# 统一 schema: 三源都对齐到这 7 列 (turnover 缺则 0; volume 腾讯缺则 0) +_UNIFIED_COLS = ["date", "open", "high", "low", "close", "volume", "turnover"] + + +def _normalize_df(df, col_map, default_turnover=0.0, default_volume=0.0): + """重命名 + 补默认列 -> _UNIFIED_COLS。缺失列用 default 兜底。""" + if df is None or len(df) == 0: + return pd.DataFrame() + df2 = df.rename(columns=col_map) + if "turnover" not in df2.columns: + df2["turnover"] = default_turnover + if "volume" not in df2.columns: + df2["volume"] = default_volume + for col in ("date", "open", "high", "low", "close"): + if col not in df2.columns: + return pd.DataFrame() # 必需列缺失 + # date 归一为 'YYYY-MM-DD' 字符串 (腾讯 date 是 date 对象, str() 即得) + df2["date"] = df2["date"].astype(str).str[:10] + return df2[_UNIFIED_COLS] + + +def fetch_sina(code): + """sina 主源 -> _UNIFIED_COLS。返空 df 表示 sina 无数据。""" + import akshare as ak + df = ak.stock_zh_index_daily(symbol=f"sh{code}") + time.sleep(SINA_SLEEP) + return _normalize_df( + df, + col_map={"date": "date", "open": "open", "high": "high", + "low": "low", "close": "close", "volume": "volume"}, + default_turnover=0.0, + ) + + +def fetch_eastmoney(code): + """东财兜底 -> _UNIFIED_COLS。VPS IP 持续 RemoteDisconnected 时返空。""" + import akshare as ak + try: + df = ak.index_zh_a_hist(symbol=code, period="daily") + except (ConnectionError, OSError, Exception) as e: + log(f" {code} eastmoney err: {type(e).__name__}: {str(e)[:80]}") + return pd.DataFrame() + time.sleep(SINA_SLEEP) + return _normalize_df( + df, + col_map={"日期": "date", "开盘": "open", "最高": "high", "最低": "low", + "收盘": "close", "成交量": "volume", "成交额": "turnover"}, + default_turnover=0.0, + ) + + +def fetch_tencent(code): + """腾讯兜底 -> _UNIFIED_COLS。 + + 实证 schema: cols=[date(date 对象), open, close, high, low, amount] — 注意: + - date 是 datetime.date 对象, str() 后取 [:10] 得 'YYYY-MM-DD' + - 无 volume 列, 用 amount (成交额) 当 turnover; volume=0 + - 列顺序 close 在 high 之前 (col_map 重命名兼容) + """ + import akshare as ak + try: + df = ak.stock_zh_index_daily_tx(symbol=f"sh{code}") + except (ConnectionError, OSError, Exception) as e: + log(f" {code} tencent err: {type(e).__name__}: {str(e)[:80]}") + return pd.DataFrame() + time.sleep(SINA_SLEEP) + return _normalize_df( + df, + col_map={"date": "date", "open": "open", "close": "close", + "high": "high", "low": "low", "amount": "turnover"}, + default_volume=0.0, + ) + + +def fetch_with_fallback(code, today): + """级联 sina → 东财 → 腾讯, 取第一个"最后日期 >= today-30"的源。 + + Returns: (df, source, warn_msg) + - df: _UNIFIED_COLS DataFrame (可能为空) + - source: 'sina'|'eastmoney'|'tencent'|'none' + - warn_msg: 兜底/降级原因 (sina 直接命中时为 None) + 全部源都 stale 时, 用最新的 stale df (有总比没有好, log WARN)。 + """ + candidates = ( + ("sina", fetch_sina), + ("eastmoney", fetch_eastmoney), + ("tencent", fetch_tencent), + ) + stale_best = None # (df, source, last_dt, reason) + warn_parts = [] + for name, fn in candidates: + try: + df = fn(code) + except Exception as e: + warn_parts.append(f"{name} 异常 {type(e).__name__}") + continue + if df is None or len(df) == 0: + warn_parts.append(f"{name} 空") + continue + try: + last_dt = pd.Timestamp(df["date"].iloc[-1]).normalize() + except Exception: + last_dt = pd.Timestamp.min + if (today - last_dt).days <= FALLBACK_RECENT_DAYS: + # 命中: 新鲜数据 + warn = None if not warn_parts else ("上游失效: " + "; ".join(warn_parts) + f" → {name} 命中") + return df, name, warn + # stale: 保留作最后兜底 + if stale_best is None or last_dt > stale_best[2]: + stale_best = (df, name, last_dt, f"{name} 停 {last_dt.date()}") + warn_parts.append(f"{name} 停 {last_dt.date()}") + + if stale_best is not None: + df, src, _, reason = stale_best + return df, src, "全源 stale (>30d); " + "; ".join(warn_parts) + return pd.DataFrame(), "none", "全源失败: " + "; ".join(warn_parts) + + +def to_dbbardata_rows(code, df): + """_UNIFIED_COLS DataFrame -> dbbardata 行元组列表 (symbol,SSE,d,...)。""" + if df is None or len(df) == 0: + return [] + out = [] + for row in df.itertuples(index=False): + dt_str = normalize_daily_dt(str(row.date)) + if not dt_str or dt_str == "None": + continue + try: + out.append(( + code, "SSE", dt_str, "d", + float(row.volume), float(row.turnover), 0.0, + float(row.open), float(row.high), + float(row.low), float(row.close), + )) + except (ValueError, TypeError) as e: + log(f" {code} skip row {dt_str}: {e}") + return out + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--lookback-days", type=int, + default=int(os.environ.get("LOOKBACK_DAYS", "30"))) + ap.add_argument("--dry-run", action="store_true") + ap.add_argument("--codes", type=str, default="", + help="逗号分隔覆盖默认 14 个 code (调试用)") + args = ap.parse_args() + + codes = (args.codes.split(",") if args.codes else CODES) + codes = [c.strip() for c in codes if c.strip()] + today = pd.Timestamp.now().normalize() + start_ts = today - pd.Timedelta(days=args.lookback_days) + start_str = start_ts.strftime("%Y-%m-%d") + log(f"start={start_str} lookback={args.lookback_days}d codes={len(codes)}" + f"{' [DRY-RUN]' if args.dry_run else ''}") + + conn = sqlite3.connect(DB, timeout=60) + conn.execute("PRAGMA busy_timeout = 60000") + conn.execute("PRAGMA journal_mode = WAL") + + ok_codes = 0 + total_rows = 0 + src_counts = {"sina": 0, "eastmoney": 0, "tencent": 0, "none": 0} + code_stats = [] # (code, source, rows, min_close, max_close, max_dt, warn?) + + conn.execute("BEGIN") + try: + for code in codes: + df, source, warn_msg = fetch_with_fallback(code, today) + src_counts[source] = src_counts.get(source, 0) + 1 + + # 过滤 lookback 范围 + if not df.empty: + df = df.copy() + df["date"] = pd.to_datetime(df["date"], format="mixed").dt.strftime("%Y-%m-%d") + df = df[df["date"] >= start_str] + + rows = to_dbbardata_rows(code, df) + if not rows: + log(f" {code} {source} 无符合 lookback 行" + + (f" ({warn_msg})" if warn_msg else "")) + code_stats.append((code, source, 0, None, None, None, + warn_msg or "lookback 范围内无行")) + continue + + if not args.dry_run: + conn.executemany( + "INSERT OR REPLACE INTO dbbardata " + "(symbol,exchange,datetime,interval,volume,turnover,open_interest," + "open_price,high_price,low_price,close_price) VALUES (?,?,?,?,?,?,?,?,?,?,?)", + rows, + ) + ok_codes += 1 + total_rows += len(rows) + closes = [r[10] for r in rows] + max_dt = max(r[2] for r in rows) + min_c = round(min(closes), 2) + max_c = round(max(closes), 2) + code_stats.append((code, source, len(rows), min_c, max_c, max_dt, warn_msg)) + log(f" {code} {source} rows={len(rows)} close=[{min_c},{max_c}] last={max_dt}" + + (f" WARN: {warn_msg}" if warn_msg else "")) + conn.execute("COMMIT") + except Exception as e: + conn.execute("ROLLBACK") + log(f"FATAL rollback: {type(e).__name__}: {e}") + conn.close() + return 1 + conn.close() + + src_summary = ", ".join(f"{k}={v}" for k, v in src_counts.items() if v) + log(f"DONE codes_ok={ok_codes}/{len(codes)} rows={total_rows} sources[{src_summary}]" + f"{' [DRY-RUN]' if args.dry_run else ''}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/data_platform/sina_index_eod_wrapper.ps1 b/scripts/data_platform/sina_index_eod_wrapper.ps1 new file mode 100644 index 0000000..1a81026 --- /dev/null +++ b/scripts/data_platform/sina_index_eod_wrapper.ps1 @@ -0,0 +1,13 @@ +# sina_index_eod_wrapper.ps1 — sanguo-idx-eod schtask wrapper (18:30 中证指数日线点位) +# 方案: 14 个中证指数 (与 SZSE 股票码碰撞) -> dbbardata('d', SSE) 隔离股票行 +# 主源 sina ak.stock_zh_index_daily, 兜底东财 ak.index_zh_a_hist +$env:http_proxy = '' +$env:https_proxy = '' +$env:all_proxy = '' +Set-Location C:\sanguo_vnpy_v2 +$ts = Get-Date -Format 'yyyyMMdd_HHmmss' +$logDir = 'C:\sanguo_vnpy_v2\data\migration_logs' +if (-not (Test-Path $logDir)) { New-Item -ItemType Directory -Path $logDir -Force | Out-Null } +$log = Join-Path $logDir "sina_idx_eod_$ts.txt" +C:\Python310\python.exe -X utf8 C:\sanguo_vnpy_v2\scripts\data_platform\sina_index_eod.py *>> $log +exit $LASTEXITCODE