feat(data): 中证指数日线点位修股票/指数同名碰撞 + 行业治偏差 section 名校正/增量接线

行情(修碰撞): 新 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 解析待专项。
2026-07-28 21:54:46 +08:00
parent c8f26bef80
commit 66a2c9e9ce
4 changed files with 334 additions and 12 deletions
@@ -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
@@ -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 等), 留作可选
 }

@@ -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())
@@ -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