1.1

check_data_details.py

@@ -0,0 +1,90 @@
+# -*- coding: utf-8 -*-
+"""检查橡胶主连 D1 OHLC 异常的细节，以及燃油主连零成交量问题。"""
+import pandas as pd
+import numpy as np
+from datetime import timezone, timedelta
+
+BJ_TZ = timezone(timedelta(hours=8))
+
+def ts_to_str(ts):
+    return pd.to_datetime(int(ts), unit='s', utc=True).tz_convert(BJ_TZ).strftime('%Y-%m-%d')
+
+print("=" * 80)
+print("橡胶主连 D1 — OHLC 异常详情")
+print("=" * 80)
+df = pd.read_parquet(r'D:\国内期货K线数据\橡胶主连_D1.parquet')
+o, h, l, c = df['open'], df['high'], df['low'], df['close']
+
+bad_h = df[h < np.maximum(o, c)]
+bad_l = df[l > np.minimum(o, c)]
+
+print(f"\nhigh < max(open, close): {len(bad_h)} 行")
+for _, r in bad_h.iterrows():
+    print(f"  {ts_to_str(r['time'])}  O={r['open']:.1f} H={r['high']:.1f} L={r['low']:.1f} C={r['close']:.1f}  (H < max(O,C)={max(r['open'],r['close']):.1f})")
+
+print(f"\nlow > min(open, close): {len(bad_l)} 行")
+for _, r in bad_l.iterrows():
+    print(f"  {ts_to_str(r['time'])}  O={r['open']:.1f} H={r['high']:.1f} L={r['low']:.1f} C={r['close']:.1f}  (L > min(O,C)={min(r['open'],r['close']):.1f})")
+
+# 检查是否是前复权导致的负价格区域
+print(f"\n前5行（最早数据）:")
+for _, r in df.head(5).iterrows():
+    print(f"  {ts_to_str(r['time'])}  O={r['open']:.1f} H={r['high']:.1f} L={r['low']:.1f} C={r['close']:.1f} V={r['tick_volume']}")
+
+print(f"\n后5行（最新数据）:")
+for _, r in df.tail(5).iterrows():
+    print(f"  {ts_to_str(r['time'])}  O={r['open']:.1f} H={r['high']:.1f} L={r['low']:.1f} C={r['close']:.1f} V={r['tick_volume']}")
+
+# 检查价格是否有负值（前复权可能导致）
+neg_price = df[df['close'] < 0]
+print(f"\n负价格行数: {len(neg_price)}")
+if len(neg_price) > 0:
+    print(neg_price[['time','open','high','low','close']].head().to_string())
+
+# 检查价格变化是否合理
+print(f"\n收盘价统计: min={df['close'].min():.1f} max={df['close'].max():.1f} mean={df['close'].mean():.1f}")
+
+# ── 燃油主连 D1 零成交量 ──
+print("\n" + "=" * 80)
+print("燃油主连 D1 — 零成交量问题")
+print("=" * 80)
+df2 = pd.read_parquet(r'D:\国内期货K线数据\燃油主连_D1.parquet')
+zero_vol = df2[df2['tick_volume'] == 0]
+print(f"总行数: {len(df2)}, 零成交量行数: {len(zero_vol)} ({len(zero_vol)/len(df2)*100:.1f}%)")
+print(f"零成交量行的时间范围: {ts_to_str(zero_vol['time'].iloc[0])} ~ {ts_to_str(zero_vol['time'].iloc[-1])}")
+# 看看零成交量是否集中在某个时间段
+if len(zero_vol) > 0:
+    zero_vol_dates = zero_vol['time'].apply(ts_to_str).values
+    years = [d[:4] for d in zero_vol_dates]
+    from collections import Counter
+    year_counts = Counter(years)
+    print(f"零成交量按年分布: {dict(sorted(year_counts.items()))}")
+
+# ── 豆二主连 D1 异常最多 ──
+print("\n" + "=" * 80)
+print("豆二主连 D1 — OHLC 异常最多")
+print("=" * 80)
+df3 = pd.read_parquet(r'D:\国内期货K线数据\豆二主连_D1.parquet')
+o3, h3, l3, c3 = df3['open'], df3['high'], df3['low'], df3['close']
+bad_h3 = df3[h3 < np.maximum(o3, c3)]
+bad_l3 = df3[l3 > np.minimum(o3, c3)]
+print(f"high < max(open,close): {len(bad_h3)} 行")
+print(f"low > min(open,close): {len(bad_l3)} 行")
+# 看看异常集中在什么时间段
+if len(bad_h3) > 0:
+    print(f"异常 high 时间范围: {ts_to_str(bad_h3['time'].iloc[0])} ~ {ts_to_str(bad_h3['time'].iloc[-1])}")
+if len(bad_l3) > 0:
+    print(f"异常 low 时间范围: {ts_to_str(bad_l3['time'].iloc[0])} ~ {ts_to_str(bad_l3['time'].iloc[-1])}")
+
+# ── 欧线集运 M15 异常涨跌 ──
+print("\n" + "=" * 80)
+print("欧线集运 M15 — 单根涨跌>50%")
+print("=" * 80)
+df4 = pd.read_parquet(r'D:\国内期货K线数据\欧线集运主连_M15.parquet')
+ret4 = df4['close'].pct_change().abs()
+huge4 = df4[ret4 > 0.5]
+for _, r in huge4.iterrows():
+    prev_close = df4.loc[df4.index.get_loc(r.name) - 1, 'close'] if df4.index.get_loc(r.name) > 0 else 0
+    print(f"  {ts_to_str(r['time'])}  prev_close={prev_close:.1f} -> close={r['close']:.1f} (ret={abs((r['close']/prev_close-1)*100) if prev_close else 0:.1f}%)")
+
+print("\n检查完成。")

check_domestic_data.py

@@ -0,0 +1,222 @@
+# -*- coding: utf-8 -*-
+"""深度检查 D:\\国内期货K线数据 目录下所有 parquet 文件的数据质量。"""
+import pandas as pd
+import numpy as np
+from pathlib import Path
+from datetime import datetime, timezone, timedelta
+import json
+
+DATA_DIR = Path(r"D:\国内期货K线数据")
+BJ_TZ = timezone(timedelta(hours=8))
+
+# 预期周期
+TIMEFRAMES = ["M15", "M30", "H1", "D1"]
+TF_SECONDS = {"M15": 900, "M30": 1800, "H1": 3600, "D1": 86400}
+
+# 预期列名
+EXPECTED_COLS = {"time", "open", "high", "low", "close", "tick_volume"}
+
+problems = []
+stats = []
+
+files = sorted(DATA_DIR.glob("*.parquet"))
+print(f"共找到 {len(files)} 个 parquet 文件\n")
+
+for f in files:
+    symbol = f.stem.rsplit("_", 1)[0]  # 如 "橡胶主连"
+    tf = f.stem.rsplit("_", 1)[1]      # 如 "D1"
+    issues = []
+
+    try:
+        df = pd.read_parquet(f)
+    except Exception as e:
+        problems.append({"file": f.name, "issues": [f"读取失败: {e}"]})
+        continue
+
+    # 1. 列名检查
+    cols = set(df.columns)
+    missing = EXPECTED_COLS - cols
+    if missing:
+        issues.append(f"缺少列: {missing}")
+
+    # 2. 行数检查
+    n = len(df)
+    if n == 0:
+        issues.append("空文件（0 行）")
+        problems.append({"file": f.name, "issues": issues})
+        continue
+    if n < 100:
+        issues.append(f"行数过少: {n} 行")
+
+    # 3. 时间戳检查
+    if "time" in df.columns:
+        ts = df["time"].values
+
+        # 检查是否有 0 或负值
+        bad_ts = np.sum(ts <= 0)
+        if bad_ts > 0:
+            issues.append(f"无效时间戳（<=0）: {bad_ts} 行")
+
+        # 检查是否升序
+        if not np.all(np.diff(ts) >= 0):
+            issues.append("时间戳未升序排列")
+
+        # 检查重复时间戳
+        dup_ts = np.sum(np.diff(ts) == 0)
+        if dup_ts > 0:
+            issues.append(f"重复时间戳: {dup_ts} 行")
+
+        # 时间范围
+        valid_ts = ts[ts > 0]
+        if len(valid_ts) > 0:
+            first_dt = pd.to_datetime(valid_ts[0], unit="s", utc=True).tz_convert(BJ_TZ)
+            last_dt = pd.to_datetime(valid_ts[-1], unit="s", utc=True).tz_convert(BJ_TZ)
+            span_days = (valid_ts[-1] - valid_ts[0]) / 86400
+
+            # 检查时间间隔是否合理
+            if len(valid_ts) > 1:
+                diffs = np.diff(valid_ts)
+                expected_gap = TF_SECONDS.get(tf, 0)
+
+                # 统计间隔异常
+                if expected_gap > 0:
+                    # 允许 5% 浮动（夜盘/午盘间隔不同）
+                    too_small = np.sum(diffs < expected_gap * 0.5)
+                    # 跨夜/跨周/跨月的大间隔是正常的
+                    # 但如果 D1 间隔 < 3600（1小时），肯定有问题
+                    if tf == "D1":
+                        too_small_d1 = np.sum(diffs < 3600)
+                        if too_small_d1 > 0:
+                            issues.append(f"D1 间隔异常小（<1h）: {too_small_d1} 处")
+                else:
+                    too_small = 0
+
+        else:
+            first_dt = last_dt = None
+            span_days = 0
+            issues.append("所有时间戳无效")
+
+    # 4. OHLC 逻辑检查
+    if all(c in df.columns for c in ["open", "high", "low", "close"]):
+        o, h, l, c = df["open"], df["high"], df["low"], df["close"]
+
+        # high >= low
+        hl_bad = np.sum((h < l).values)
+        if hl_bad > 0:
+            issues.append(f"high < low: {hl_bad} 行")
+
+        # high >= max(open, close)
+        hc_bad = np.sum((h < np.maximum(o, c)).values)
+        if hc_bad > 0:
+            issues.append(f"high < max(open,close): {hc_bad} 行")
+
+        # low <= min(open, close)
+        lc_bad = np.sum((l > np.minimum(o, c)).values)
+        if lc_bad > 0:
+            issues.append(f"low > min(open,close): {lc_bad} 行")
+
+    # 5. 价格合理性检查
+    if "close" in df.columns:
+        close = df["close"]
+        # 负价格
+        neg = np.sum((close < 0).values)
+        if neg > 0:
+            issues.append(f"负价格: {neg} 行")
+
+        # 零价格
+        zero = np.sum((close == 0).values)
+        if zero > 0:
+            issues.append(f"零价格: {zero} 行")
+
+        # 异常大的价格变动（单根 bar 涨跌超过 50%）
+        if len(close) > 1:
+            ret = close.pct_change().abs()
+            huge = np.sum((ret > 0.5).values)
+            if huge > 0:
+                issues.append(f"单根 bar 涨跌>50%: {huge} 处")
+
+    # 6. 成交量检查
+    if "tick_volume" in df.columns:
+        vol = df["tick_volume"]
+        neg_vol = np.sum((vol < 0).values)
+        if neg_vol > 0:
+            issues.append(f"负成交量: {neg_vol} 行")
+        zero_vol = np.sum((vol == 0).values)
+        if zero_vol > 0:
+            # volume=0 可能是非交易时段残留，记录但不一定是严重问题
+            pct = zero_vol / len(df) * 100
+            if pct > 5:
+                issues.append(f"零成交量占比 {pct:.1f}% ({zero_vol} 行)")
+
+    # 7. 数据覆盖率检查 — 各周期应有足够历史
+    if span_days > 0:
+        if tf == "D1" and span_days < 365:
+            issues.append(f"D1 历史不足1年: {span_days:.0f} 天")
+        elif tf == "H1" and span_days < 180:
+            issues.append(f"H1 历史不足半年: {span_days:.0f} 天")
+        elif tf == "M15" and span_days < 90:
+            issues.append(f"M15 历史不足3月: {span_days:.0f} 天")
+        elif tf == "M30" and span_days < 90:
+            issues.append(f"M30 历史不足3月: {span_days:.0f} 天")
+
+    stat = {
+        "file": f.name,
+        "symbol": symbol,
+        "tf": tf,
+        "rows": n,
+        "first": str(first_dt) if first_dt else "N/A",
+        "last": str(last_dt) if last_dt else "N/A",
+        "span_days": round(span_days, 1) if span_days else 0,
+        "issues": issues if issues else "OK",
+    }
+    stats.append(stat)
+    if issues:
+        problems.append({"file": f.name, "issues": issues})
+
+# ── 汇总输出 ──
+print("=" * 80)
+print("数据质量深度检查报告")
+print("=" * 80)
+print(f"检查时间: {datetime.now(BJ_TZ).strftime('%Y-%m-%d %H:%M:%S')}")
+print(f"总文件数: {len(files)}")
+print(f"有问题文件: {len(problems)}")
+print(f"正常文件: {len(files) - len(problems)}")
+print()
+
+if problems:
+    print("-" * 80)
+    print("问题详情:")
+    print("-" * 80)
+    for p in problems:
+        print(f"\n  [{p['file']}]")
+        for iss in p["issues"]:
+            print(f"    ❌ {iss}")
+else:
+    print("✅ 所有文件检查通过，无异常！")
+
+# 按品种汇总
+print("\n" + "=" * 80)
+print("按品种汇总:")
+print("=" * 80)
+symbols = sorted(set(s["symbol"] for s in stats))
+print(f"{'品种':<12} {'周期':<5} {'行数':>6} {'起始时间':<22} {'结束时间':<22} {'跨度(天)':>8} {'状态':<8}")
+print("-" * 90)
+for sym in symbols:
+    for tf in TIMEFRAMES:
+        matches = [s for s in stats if s["symbol"] == sym and s["tf"] == tf]
+        if matches:
+            s = matches[0]
+            status = "❌" if s["issues"] != "OK" else "✅"
+            print(f"{sym:<12} {tf:<5} {s['rows']:>6} {s['first']:<22} {s['last']:<22} {s['span_days']:>8} {status:<8}")
+
+# 按周期统计行数范围
+print("\n" + "=" * 80)
+print("各周期行数统计:")
+print("=" * 80)
+for tf in TIMEFRAMES:
+    tf_stats = [s for s in stats if s["tf"] == tf]
+    if tf_stats:
+        rows = [s["rows"] for s in tf_stats]
+        print(f"  {tf}: 品种数={len(tf_stats)}, 行数范围={min(rows)}~{max(rows)}, 平均={int(np.mean(rows))}")
+
+print("\n检查完成！")

deep_check_all.py

@@ -0,0 +1,165 @@
+# -*- coding: utf-8 -*-
+"""深度检查 D:\国内期货K线数据 下全部 parquet 文件的数据质量。"""
+import os
+import sys
+from pathlib import Path
+import pandas as pd
+import numpy as np
+from datetime import timezone, timedelta
+
+BJ_TZ = timezone(timedelta(hours=8))
+DATA_DIR = Path(r"D:\国内期货K线数据")
+
+# 全部 parquet 文件
+files = sorted(DATA_DIR.glob("*.parquet"))
+print(f"找到 {len(files)} 个 parquet 文件")
+print("=" * 100)
+
+# 汇总统计
+total_files = 0
+ok_files = 0
+problem_files = 0
+problems = []  # (filename, issue_type, detail)
+
+# 按品种分组统计
+symbol_stats = {}  # {symbol: {tf: {rows, first, last, bad_h, bad_l, neg, zero_vol, huge_ret}}}
+
+for f in files:
+    total_files += 1
+    name = f.stem  # e.g. 橡胶主连_D1
+    parts = name.rsplit("_", 1)
+    if len(parts) == 2:
+        symbol, tf = parts
+    else:
+        symbol, tf = name, "?"
+
+    try:
+        df = pd.read_parquet(f)
+    except Exception as e:
+        problems.append((name, "READ_ERROR", str(e)))
+        problem_files += 1
+        continue
+
+    rows = len(df)
+    if rows == 0:
+        problems.append((name, "EMPTY", "0 rows"))
+        problem_files += 1
+        continue
+
+    o = df["open"].values
+    h = df["high"].values
+    l = df["low"].values
+    c = df["close"].values
+    v = df["tick_volume"].values
+
+    # OHLC 逻辑检查
+    bad_h = int(np.sum((h < o) | (h < c)))
+    bad_l = int(np.sum((l > o) | (l > c)))
+    neg_price = int(np.sum(c < 0))
+    zero_vol = int(np.sum(v == 0))
+
+    # 异常涨跌 (>50%)
+    if rows > 1:
+        rets = np.abs(np.diff(c) / c[:-1])
+        huge_ret = int(np.sum(rets > 0.5))
+    else:
+        huge_ret = 0
+
+    # 时间范围
+    first_ts = int(df["time"].iloc[0])
+    last_ts = int(df["time"].iloc[-1])
+    first_dt = pd.to_datetime(first_ts, unit="s", utc=True).tz_convert(BJ_TZ).strftime("%Y-%m-%d")
+    last_dt = pd.to_datetime(last_ts, unit="s", utc=True).tz_convert(BJ_TZ).strftime("%Y-%m-%d")
+
+    # 记录统计
+    if symbol not in symbol_stats:
+        symbol_stats[symbol] = {}
+    symbol_stats[symbol][tf] = {
+        "rows": rows,
+        "first": first_dt,
+        "last": last_dt,
+        "bad_h": bad_h,
+        "bad_l": bad_l,
+        "neg": neg_price,
+        "zero_vol": zero_vol,
+        "huge_ret": huge_ret,
+    }
+
+    has_problem = bad_h > 0 or bad_l > 0 or neg_price > 0 or zero_vol > 0 or huge_ret > 0
+    if has_problem:
+        problem_files += 1
+        if bad_h > 0:
+            problems.append((name, "OHLC_BAD_H", f"{bad_h} rows"))
+        if bad_l > 0:
+            problems.append((name, "OHLC_BAD_L", f"{bad_l} rows"))
+        if neg_price > 0:
+            problems.append((name, "NEG_PRICE", f"{neg_price} rows"))
+        if zero_vol > 0:
+            problems.append((name, "ZERO_VOL", f"{zero_vol} rows ({zero_vol/rows*100:.1f}%)"))
+        if huge_ret > 0:
+            problems.append((name, "HUGE_RET", f"{huge_ret} rows (>50%)"))
+    else:
+        ok_files += 1
+
+# ── 输出结果 ──
+print(f"\n总计: {total_files} 文件, OK: {ok_files}, 有问题: {problem_files}")
+print("=" * 100)
+
+if problems:
+    print("\n=== 问题清单 ===")
+    for name, issue, detail in problems:
+        print(f"  {name:30s}  {issue:15s}  {detail}")
+else:
+    print("\n=== 全部文件通过质量检查，无任何问题！ ===")
+
+# ── 按品种汇总 ──
+print("\n" + "=" * 100)
+print("按品种汇总 (只显示有问题的品种)")
+print("=" * 100)
+problem_symbols = set()
+for sym, tfs in symbol_stats.items():
+    for tf, s in tfs.items():
+        if s["bad_h"] or s["bad_l"] or s["neg"] or s["zero_vol"] or s["huge_ret"]:
+            problem_symbols.add(sym)
+
+if problem_symbols:
+    for sym in sorted(problem_symbols):
+        print(f"\n{sym}:")
+        for tf in ["M1", "M3", "M5", "M15", "M30", "H1", "H2", "H4", "D1", "W1"]:
+            if tf in symbol_stats[sym]:
+                s = symbol_stats[sym][tf]
+                issues = []
+                if s["bad_h"]: issues.append(f"bad_h={s['bad_h']}")
+                if s["bad_l"]: issues.append(f"bad_l={s['bad_l']}")
+                if s["neg"]: issues.append(f"neg={s['neg']}")
+                if s["zero_vol"]: issues.append(f"zero_vol={s['zero_vol']}")
+                if s["huge_ret"]: issues.append(f"huge_ret={s['huge_ret']}")
+                if issues:
+                    print(f"  {tf}: rows={s['rows']}, {', '.join(issues)}")
+else:
+    print("无问题品种！")
+
+# ── 全部品种周期一览表 ──
+print("\n" + "=" * 100)
+print("全部品种 × 周期 一览表")
+print("=" * 100)
+header = f"{'品种':<14s} |"
+for tf in ["M1", "M3", "M5", "M15", "M30", "H1", "H2", "H4", "D1", "W1"]:
+    header += f" {tf:>6s} |"
+print(header)
+print("-" * len(header))
+
+for sym in sorted(symbol_stats.keys()):
+    row = f"{sym:<14s} |"
+    for tf in ["M1", "M3", "M5", "M15", "M30", "H1", "H2", "H4", "D1", "W1"]:
+        if tf in symbol_stats[sym]:
+            s = symbol_stats[sym][tf]
+            if s["bad_h"] or s["bad_l"] or s["neg"] or s["zero_vol"] or s["huge_ret"]:
+                row += f" {'BAD':>6s} |"
+            else:
+                row += f" {s['rows']:>6d} |"
+        else:
+            row += f" {'---':>6s} |"
+    print(row)
+
+print(f"\n总计: {total_files} 文件, OK: {ok_files}, 有问题: {problem_files}")

deep_investigate.py

@@ -0,0 +1,131 @@
+# -*- coding: utf-8 -*-
+"""深入排查校验发现的问题。
+
+1. 旧数据残留: 已保存文件比重新拉取多出的 bar
+2. D1 时间间隔异常: 看看具体是哪些日期，是否都是节假日
+"""
+import pandas as pd
+import numpy as np
+from pathlib import Path
+from datetime import timezone, timedelta
+
+BJ_TZ = timezone(timedelta(hours=8))
+DATA_DIR = Path(r"D:\国内期货K线数据")
+
+# ── 1. 旧数据残留排查 ──
+print("=" * 100)
+print("1. 旧数据残留排查: 检查全部 600 个文件是否有重复或异常时间戳")
+print("=" * 100)
+
+# 之前校验发现沪铜M15多出 2025-03-18 00:00/00:15/00:30
+# 这些是夜盘时间段，但 tqsdk 新拉取没有 — 可能是旧数据残留
+# 检查所有 M15 文件的夜盘异常时间
+
+night_issues = []
+for f in sorted(DATA_DIR.glob("*_M15.parquet")):
+    name = f.stem
+    df = pd.read_parquet(f)
+    # 提取每个 bar 的小时分钟
+    dt_list = [pd.to_datetime(int(ts), unit="s", utc=True).tz_convert(BJ_TZ) for ts in df["time"].values]
+    hours = np.array([dt.hour for dt in dt_list])
+    # 夜盘时间应该是 21:00-23:59 + 09:00-15:15
+    # 异常: 00:00-09:00 之间的 M15 bar（除了 09:00 本身）
+    # 或者 15:30-21:00 之间的 bar
+    abnormal = ((hours > 0) & (hours < 9)) | ((hours >= 15) & (hours < 21) & (hours != 0))
+    abnormal_count = abnormal.sum()
+    if abnormal_count > 0:
+        abnormal_times = [dt for dt, ab in zip(dt_list, abnormal) if ab]
+        abnormal_strs = [dt.strftime("%Y-%m-%d %H:%M") for dt in abnormal_times[:5]]
+        night_issues.append((name, abnormal_count, abnormal_strs))
+        print(f"  {name}: {abnormal_count} 条异常时间 bar, 例: {abnormal_strs}")
+
+if not night_issues:
+    print("  无异常时间 bar")
+
+# ── 2. D1 时间间隔异常排查 ──
+print("\n" + "=" * 100)
+print("2. D1 时间间隔异常排查: 橡胶主连 D1 的异常间隔")
+print("=" * 100)
+
+df = pd.read_parquet(DATA_DIR / "橡胶主连_D1.parquet")
+times = df["time"].values
+diffs = np.diff(times)
+
+# 统计各种间隔
+from collections import Counter
+diff_days = (diffs / 86400).astype(int)
+diff_counter = Counter(diff_days)
+print("间隔天数分布:")
+for d, cnt in sorted(diff_counter.items()):
+    label = ""
+    if d == 1: label = " (正常: 次日)"
+    elif d == 2: label = " (周末: 周五→周一)"
+    elif d == 3: label = " (长假3天)"
+    elif d == 4: label = " (长假4天)"
+    elif d == 5: label = " (长假5天)"
+    elif d == 7: label = " (长假7天)"
+    elif d > 7: label = " (超长间隔!)"
+    print(f"  {d} 天: {cnt} 次{label}")
+
+# 看看异常的具体是哪些日期
+print("\n异常间隔的具体日期 (>3天):")
+for i, d in enumerate(diff_days):
+    if d > 3:
+        from_dt = pd.to_datetime(int(times[i]), unit="s", utc=True).tz_convert(BJ_TZ).strftime("%Y-%m-%d")
+        to_dt = pd.to_datetime(int(times[i+1]), unit="s", utc=True).tz_convert(BJ_TZ).strftime("%Y-%m-%d")
+        print(f"  {from_dt} → {to_dt}  (间隔 {d} 天)")
+
+# ── 3. 检查残留 bar 的具体情况 ──
+print("\n" + "=" * 100)
+print("3. 残留 bar 具体情况: 沪铜主连 M15 的多出 bar")
+print("=" * 100)
+
+# 读沪铜 M15
+df = pd.read_parquet(DATA_DIR / "沪铜主连_M15.parquet")
+# 看 2025-03-18 附近的数据
+mask = (df["time"] >= pd.Timestamp("2025-03-17", tz="UTC").timestamp()) & \
+       (df["time"] <= pd.Timestamp("2025-03-19", tz="UTC").timestamp())
+sub = df[mask]
+print("沪铜主连 M15 在 2025-03-17 ~ 2025-03-19 附近的数据:")
+for _, r in sub.iterrows():
+    dt = pd.to_datetime(int(r["time"]), unit="s", utc=True).tz_convert(BJ_TZ).strftime("%Y-%m-%d %H:%M")
+    print(f"  {dt}  O={r['open']:.0f} H={r['high']:.0f} L={r['low']:.0f} C={r['close']:.0f} V={int(r['tick_volume'])}")
+
+# ── 4. 涨跌>30% 排查 ──
+print("\n" + "=" * 100)
+print("4. 涨跌>30% 排查: 欧线集运 D1")
+print("=" * 100)
+
+df = pd.read_parquet(DATA_DIR / "欧线集运主连_D1.parquet")
+rets = df["close"].pct_change().abs()
+huge = df[rets > 0.3]
+print(f"欧线集运 D1 涨跌>30%: {len(huge)} 条")
+for _, r in huge.iterrows():
+    idx = df.index.get_loc(r.name)
+    if idx > 0:
+        prev = df.iloc[idx - 1]
+        dt = pd.to_datetime(int(r["time"]), unit="s", utc=True).tz_convert(BJ_TZ).strftime("%Y-%m-%d")
+        prev_dt = pd.to_datetime(int(prev["time"]), unit="s", utc=True).tz_convert(BJ_TZ).strftime("%Y-%m-%d")
+        ret = (r["close"] / prev["close"] - 1) * 100
+        print(f"  {prev_dt} C={prev['close']:.0f} → {dt} C={r['close']:.0f}  ({ret:+.1f}%)")
+
+# ── 5. 燃油主连 D1 异常间隔最多(57条) ──
+print("\n" + "=" * 100)
+print("5. 燃油主连 D1 异常间隔 (57条)")
+print("=" * 100)
+
+df = pd.read_parquet(DATA_DIR / "燃油主连_D1.parquet")
+times = df["time"].values
+diffs = np.diff(times)
+diff_days = (diffs / 86400).astype(int)
+diff_counter = Counter(diff_days)
+print("间隔天数分布:")
+for d, cnt in sorted(diff_counter.items()):
+    print(f"  {d} 天: {cnt} 次")
+
+print("\n燃油 D1 异常间隔 (>3天) 的具体日期:")
+for i, d in enumerate(diff_days):
+    if d > 3:
+        from_dt = pd.to_datetime(int(times[i]), unit="s", utc=True).tz_convert(BJ_TZ).strftime("%Y-%m-%d")
+        to_dt = pd.to_datetime(int(times[i+1]), unit="s", utc=True).tz_convert(BJ_TZ).strftime("%Y-%m-%d")
+        print(f"  {from_dt} → {to_dt}  (间隔 {d} 天)")

fetch_all_domestic_futures.py

@@ -0,0 +1,149 @@
+# -*- coding: utf-8 -*-
+"""批量拉取全部 60 个国内期货品种 × 10 个周期的主连数据。
+
+tqsdk data_length=10000（上限），前复权 adj_type="F"。
+每个品种每个周期建连接→拉数据→关连接，间隔 0.3s。
+"""
+import json
+import time
+import sys
+from pathlib import Path
+
+import pandas as pd
+from datetime import timezone, timedelta
+from tqsdk import TqApi, TqAuth, TqSim
+
+# ── 配置 ──
+settings = json.loads(Path(r"D:\cl\AlphaMaster\web_settings.json").read_text(encoding="utf-8"))
+user = settings.get("tqsdk_user", "")
+pwd = settings.get("tqsdk_password", "")
+if not user or not pwd:
+    user, pwd = "七斗居士", "ghhkphs8"
+
+DATA_LENGTH = 10000
+BJ_TZ = timezone(timedelta(hours=8))
+OUT_DIR = Path(r"D:\国内期货K线数据")
+OUT_DIR.mkdir(parents=True, exist_ok=True)
+
+# 全部 60 品种映射
+SYMBOL_MAP = {
+    "沪铜主连": "KQ.m@SHFE.cu", "沪铝主连": "KQ.m@SHFE.al", "沪锌主连": "KQ.m@SHFE.zn",
+    "沪铅主连": "KQ.m@SHFE.pb", "沪镍主连": "KQ.m@SHFE.ni", "沪锡主连": "KQ.m@SHFE.sn",
+    "沪金主连": "KQ.m@SHFE.au", "沪银主连": "KQ.m@SHFE.ag", "螺纹钢主连": "KQ.m@SHFE.rb",
+    "热卷主连": "KQ.m@SHFE.hc", "橡胶主连": "KQ.m@SHFE.ru", "燃油主连": "KQ.m@SHFE.fu",
+    "沥青主连": "KQ.m@SHFE.bu", "纸浆主连": "KQ.m@SHFE.sp", "不锈钢主连": "KQ.m@SHFE.ss",
+    "铁矿石主连": "KQ.m@DCE.i", "焦炭主连": "KQ.m@DCE.j", "焦煤主连": "KQ.m@DCE.jm",
+    "豆粕主连": "KQ.m@DCE.m", "豆油主连": "KQ.m@DCE.y", "棕榈油主连": "KQ.m@DCE.p",
+    "玉米主连": "KQ.m@DCE.c", "淀粉主连": "KQ.m@DCE.cs", "豆一主连": "KQ.m@DCE.a",
+    "豆二主连": "KQ.m@DCE.b", "塑料主连": "KQ.m@DCE.l", "PVC主连": "KQ.m@DCE.v",
+    "聚丙烯主连": "KQ.m@DCE.pp", "LPG主连": "KQ.m@DCE.pg", "乙二醇主连": "KQ.m@DCE.eg",
+    "苯乙烯主连": "KQ.m@DCE.eb",
+    "白糖主连": "KQ.m@CZCE.SR", "棉花主连": "KQ.m@CZCE.CF", "PTA主连": "KQ.m@CZCE.TA",
+    "甲醇主连": "KQ.m@CZCE.MA", "菜油主连": "KQ.m@CZCE.OI", "菜粕主连": "KQ.m@CZCE.RM",
+    "玻璃主连": "KQ.m@CZCE.FG", "纯碱主连": "KQ.m@CZCE.SA", "硅铁主连": "KQ.m@CZCE.SF",
+    "锰硅主连": "KQ.m@CZCE.SM", "苹果主连": "KQ.m@CZCE.AP", "红枣主连": "KQ.m@CZCE.CJ",
+    "尿素主连": "KQ.m@CZCE.UR", "烧碱主连": "KQ.m@CZCE.SH",
+    "沪深300主连": "KQ.m@CFFEX.IF", "中证500主连": "KQ.m@CFFEX.IC",
+    "上证50主连": "KQ.m@CFFEX.IH", "中证1000主连": "KQ.m@CFFEX.IM",
+    "10年国债主连": "KQ.m@CFFEX.T", "5年国债主连": "KQ.m@CFFEX.TF",
+    "2年国债主连": "KQ.m@CFFEX.TS", "30年国债主连": "KQ.m@CFFEX.TL",
+    "原油主连": "KQ.m@INE.sc", "低硫燃油主连": "KQ.m@INE.lu",
+    "20号胶主连": "KQ.m@INE.nr", "国际铜主连": "KQ.m@INE.bc",
+    "欧线集运主连": "KQ.m@INE.ec",
+    "工业硅主连": "KQ.m@GFEX.si", "碳酸锂主连": "KQ.m@GFEX.lc",
+}
+
+# 10 个周期
+TIMEFRAMES = [
+    (60,      "M1"),
+    (180,     "M3"),
+    (300,     "M5"),
+    (900,     "M15"),
+    (1800,    "M30"),
+    (3600,    "H1"),
+    (7200,    "H2"),
+    (14400,   "H4"),
+    (86400,   "D1"),
+    (604800,  "W1"),
+]
+
+total = len(SYMBOL_MAP) * len(TIMEFRAMES)
+done = 0
+ok = 0
+fail = 0
+skipped = 0
+
+print(f"tqsdk: {user}")
+print(f"品种: {len(SYMBOL_MAP)} 个 × 周期: {len(TIMEFRAMES)} 个 = {total} 个文件")
+print("=" * 80)
+sys.stdout.flush()
+
+for sym_name, tqsdk_sym in SYMBOL_MAP.items():
+    for dur, tf_name in TIMEFRAMES:
+        done += 1
+        out_file = OUT_DIR / f"{sym_name}_{tf_name}.parquet"
+
+        # 跳过已存在的文件（避免重复拉取，除非用 --force）
+        # 这里不跳过，全部重新拉取
+
+        api = None
+        df_raw = None
+        try:
+            api = TqApi(TqSim(), auth=TqAuth(user, pwd), disable_print=True)
+            df_raw = api.get_kline_serial(tqsdk_sym, dur, data_length=DATA_LENGTH, adj_type="F")
+        except Exception as exc:
+            print(f"[{done}/{total}] {sym_name} {tf_name}: ERROR {exc}")
+            fail += 1
+            sys.stdout.flush()
+            continue
+        finally:
+            if api is not None:
+                try: api.close()
+                except Exception: pass
+
+        if df_raw is None or df_raw.empty:
+            print(f"[{done}/{total}] {sym_name} {tf_name}: NO DATA")
+            fail += 1
+            sys.stdout.flush()
+            time.sleep(0.3)
+            continue
+
+        # 过滤
+        df = df_raw[df_raw["datetime"] != 0].copy()
+        df = df[df["volume"] > 0].copy()
+
+        if len(df) == 0:
+            print(f"[{done}/{total}] {sym_name} {tf_name}: EMPTY after filter")
+            fail += 1
+            sys.stdout.flush()
+            time.sleep(0.3)
+            continue
+
+        # 转换
+        result = pd.DataFrame()
+        result["time"] = (df["datetime"].astype("int64") // 1_000_000_000).astype("int64")
+        result["open"] = df["open"].astype("float64")
+        result["high"] = df["high"].astype("float64")
+        result["low"] = df["low"].astype("float64")
+        result["close"] = df["close"].astype("float64")
+        result["tick_volume"] = df["volume"].astype("int64")
+        result = result.sort_values("time").reset_index(drop=True)
+        result = result.drop_duplicates(subset=["time"], keep="last").reset_index(drop=True)
+
+        result.to_parquet(out_file, index=False)
+
+        first = pd.to_datetime(int(result["time"].iloc[0]), unit="s", utc=True).tz_convert(BJ_TZ).strftime("%Y-%m-%d")
+        last = pd.to_datetime(int(result["time"].iloc[-1]), unit="s", utc=True).tz_convert(BJ_TZ).strftime("%m-%d")
+        o, h, l, c = result["open"], result["high"], result["low"], result["close"]
+        bad = int(((h < o) | (h < c) | (l > o) | (l > c)).sum())
+        print(f"[{done}/{total}] {sym_name} {tf_name}: {len(result)} rows, {first}~{last}, bad={bad}")
+        sys.stdout.flush()
+        ok += 1
+        time.sleep(0.3)
+
+    # 每个品种完成后短暂休息
+    time.sleep(0.5)
+
+print("\n" + "=" * 80)
+print(f"完成: total={total} ok={ok} fail={fail}")
+print("=" * 80)

fetch_rubber_data.py

@@ -0,0 +1,137 @@
+# -*- coding: utf-8 -*-
+"""拉取橡胶主连各时间周期数据并保存到 D:\国内期货K线数据。
+
+使用 tqsdk 天勤量化，前复权 adj_type="F"。
+周期: M15, M30, H1, D1
+"""
+import json
+import sys
+import time
+from pathlib import Path
+
+import pandas as pd
+from datetime import timezone, timedelta
+
+# 读取 tqsdk 凭证
+settings_path = Path(r"D:\cl\AlphaMaster\web_settings.json")
+try:
+    settings = json.loads(settings_path.read_text(encoding="utf-8"))
+    user = str(settings.get("tqsdk_user", "")).strip()
+    pwd = str(settings.get("tqsdk_password", "")).strip()
+except Exception:
+    user, pwd = "", ""
+
+if not user or not pwd:
+    user, pwd = "七斗居士", "ghhkphs8"
+
+print(f"tqsdk 账号: {user}")
+print("=" * 60)
+
+# 橡胶主连 tqsdk 代码
+TQSDK_SYMBOL = "KQ.m@SHFE.ru"
+
+# 周期映射: tqsdk duration_seconds
+TIMEFRAMES = {
+    "M15": 900,
+    "M30": 1800,
+    "H1": 3600,
+    "D1": 86400,
+}
+
+# 数据长度（tqsdk 最多支持 8000+ 根）
+DATA_LENGTH = 8000
+
+BJ_TZ = timezone(timedelta(hours=8))
+
+out_dir = Path(r"D:\国内期货K线数据")
+out_dir.mkdir(parents=True, exist_ok=True)
+
+from tqsdk import TqApi, TqAuth, TqSim
+
+for tf_name, duration in TIMEFRAMES.items():
+    print(f"\n正在拉取 橡胶主连 {tf_name}...")
+    api = None
+    try:
+        api = TqApi(TqSim(), auth=TqAuth(user, pwd), disable_print=True)
+        # adj_type="F" 前复权
+        df = api.get_kline_serial(
+            TQSDK_SYMBOL, duration, data_length=DATA_LENGTH, adj_type="F"
+        )
+        print(f"  拉取到 {len(df)} 根 K 线")
+    except Exception as exc:
+        print(f"  [ERROR] tqsdk 连接失败: {exc}")
+        if api is not None:
+            try:
+                api.close()
+            except Exception:
+                pass
+        continue
+    finally:
+        if api is not None:
+            try:
+                api.close()
+            except Exception:
+                pass
+
+    if df is None or df.empty:
+        print(f"  [ERROR] 未拉取到数据")
+        continue
+
+    # 过滤无效行
+    df = df[df["datetime"] != 0].copy()
+    df = df[df["volume"] > 0].copy()
+    print(f"  过滤后有效 K 线: {len(df)} 根")
+
+    if len(df) == 0:
+        print(f"  [ERROR] 过滤后无有效数据")
+        continue
+
+    # 转换为项目兼容格式
+    result = pd.DataFrame()
+    result["time"] = (df["datetime"].astype("int64") // 1_000_000_000).astype("int64")
+    result["open"] = df["open"].astype("float64")
+    result["high"] = df["high"].astype("float64")
+    result["low"] = df["low"].astype("float64")
+    result["close"] = df["close"].astype("float64")
+    result["tick_volume"] = df["volume"].astype("int64")
+
+    # 升序排列 + 去重
+    result = result.sort_values("time").reset_index(drop=True)
+    result = result.drop_duplicates(subset=["time"], keep="last").reset_index(drop=True)
+
+    # 保存
+    out_file = out_dir / f"橡胶主连_{tf_name}.parquet"
+    result.to_parquet(out_file, index=False)
+
+    # 统计
+    first_dt = pd.to_datetime(result["time"].iloc[0], unit="s", utc=True).tz_convert(BJ_TZ)
+    last_dt = pd.to_datetime(result["time"].iloc[-1], unit="s", utc=True).tz_convert(BJ_TZ)
+    span_days = (result["time"].iloc[-1] - result["time"].iloc[0]) / 86400
+
+    print(f"  保存: {out_file}")
+    print(f"  总行数: {len(result)}")
+    print(f"  时间范围: {first_dt.strftime('%Y-%m-%d %H:%M')} ~ {last_dt.strftime('%Y-%m-%d %H:%M')}")
+    print(f"  跨度: {span_days:.0f} 天 ({span_days/365.25:.1f} 年)")
+    print(f"  收盘价: {result['close'].min():.1f} ~ {result['close'].max():.1f}")
+    print(f"  前3行:")
+    for _, r in result.head(3).iterrows():
+        dt = pd.to_datetime(int(r['time']), unit='s', utc=True).tz_convert(BJ_TZ).strftime('%Y-%m-%d %H:%M')
+        print(f"    {dt}  O={r['open']:.1f} H={r['high']:.1f} L={r['low']:.1f} C={r['close']:.1f} V={int(r['tick_volume'])}")
+    print(f"  后3行:")
+    for _, r in result.tail(3).iterrows():
+        dt = pd.to_datetime(int(r['time']), unit='s', utc=True).tz_convert(BJ_TZ).strftime('%Y-%m-%d %H:%M')
+        print(f"    {dt}  O={r['open']:.1f} H={r['high']:.1f} L={r['low']:.1f} C={r['close']:.1f} V={int(r['tick_volume'])}")
+
+    # 简单质量检查
+    o, h, l, c = result["open"], result["high"], result["low"], result["close"]
+    bad_h = ((h < o) | (h < c)).sum()
+    bad_l = ((l > o) | (l > c)).sum()
+    neg = (c < 0).sum()
+    zero_vol = (result["tick_volume"] == 0).sum()
+    print(f"  质量检查: bad_h={bad_h} bad_l={bad_l} neg_price={neg} zero_vol={zero_vol}")
+
+    time.sleep(1)  # 避免频繁连接
+
+print("\n" + "=" * 60)
+print("全部拉取完成!")
+print("=" * 60)

fetch_rubber_full.py

@@ -0,0 +1,124 @@
+# -*- coding: utf-8 -*-
+"""拉取橡胶主连全部周期数据，data_length=10000（tqsdk 上限）。
+
+覆盖周期: 1m, 3m, 5m, 15m, 30m, 1h, 2h, 4h, 1d, 1w
+保存路径: D:\\国内期货K线数据\\橡胶主连_{TF}.parquet
+"""
+import json
+import time
+from pathlib import Path
+
+import pandas as pd
+from datetime import timezone, timedelta
+from tqsdk import TqApi, TqAuth, TqSim
+
+# ── 配置 ──
+settings = json.loads(Path(r"D:\cl\AlphaMaster\web_settings.json").read_text(encoding="utf-8"))
+user = settings.get("tqsdk_user", "")
+pwd = settings.get("tqsdk_password", "")
+if not user or not pwd:
+    user, pwd = "七斗居士", "ghhkphs8"
+
+TQSDK_SYMBOL = "KQ.m@SHFE.ru"  # 橡胶主连
+DATA_LENGTH = 10000  # tqsdk 上限
+BJ_TZ = timezone(timedelta(hours=8))
+OUT_DIR = Path(r"D:\国内期货K线数据")
+
+# 全部支持的周期: (duration_seconds, 文件名后缀, 显示名)
+TIMEFRAMES = [
+    (60,      "M1",  "1分钟"),
+    (180,     "M3",  "3分钟"),
+    (300,     "M5",  "5分钟"),
+    (900,     "M15", "15分钟"),
+    (1800,    "M30", "30分钟"),
+    (3600,    "H1",  "1小时"),
+    (7200,    "H2",  "2小时"),
+    (14400,   "H4",  "4小时"),
+    (86400,   "D1",  "日线"),
+    (604800,  "W1",  "周线"),
+]
+
+print(f"tqsdk 账号: {user}")
+print(f"品种: 橡胶主连 ({TQSDK_SYMBOL})")
+print(f"数据长度上限: {DATA_LENGTH}")
+print("=" * 80)
+
+for dur, tf_name, tf_label in TIMEFRAMES:
+    print(f"\n正在拉取 橡胶主连 {tf_label} ({tf_name})...")
+    api = None
+    df = None
+    try:
+        api = TqApi(TqSim(), auth=TqAuth(user, pwd), disable_print=True)
+        df = api.get_kline_serial(TQSDK_SYMBOL, dur, data_length=DATA_LENGTH, adj_type="F")
+        print(f"  拉取到 {len(df)} 根 K 线")
+    except Exception as exc:
+        print(f"  [ERROR] {exc}")
+    finally:
+        if api is not None:
+            try: api.close()
+            except Exception: pass
+
+    if df is None or df.empty:
+        print(f"  [SKIP] 无数据")
+        continue
+
+    # 过滤无效行
+    before = len(df)
+    df = df[df["datetime"] != 0].copy()
+    df = df[df["volume"] > 0].copy()
+    after = len(df)
+    print(f"  过滤: {before} -> {after} 根 (去除 {before - after} 根无效)")
+
+    if after == 0:
+        print(f"  [SKIP] 过滤后无数据")
+        continue
+
+    # 转换格式
+    result = pd.DataFrame()
+    result["time"] = (df["datetime"].astype("int64") // 1_000_000_000).astype("int64")
+    result["open"] = df["open"].astype("float64")
+    result["high"] = df["high"].astype("float64")
+    result["low"] = df["low"].astype("float64")
+    result["close"] = df["close"].astype("float64")
+    result["tick_volume"] = df["volume"].astype("int64")
+
+    result = result.sort_values("time").reset_index(drop=True)
+    result = result.drop_duplicates(subset=["time"], keep="last").reset_index(drop=True)
+
+    out_file = OUT_DIR / f"橡胶主连_{tf_name}.parquet"
+    result.to_parquet(out_file, index=False)
+
+    # 统计
+    first_dt = pd.to_datetime(int(result["time"].iloc[0]), unit="s", utc=True).tz_convert(BJ_TZ)
+    last_dt = pd.to_datetime(int(result["time"].iloc[-1]), unit="s", utc=True).tz_convert(BJ_TZ)
+    span_days = (int(result["time"].iloc[-1]) - int(result["time"].iloc[0])) / 86400
+
+    # 质量检查
+    o, h, l, c = result["open"], result["high"], result["low"], result["close"]
+    bad_h = int(((h < o) | (h < c)).sum())
+    bad_l = int(((l > o) | (l > c)).sum())
+    neg = int((c < 0).sum())
+    zero_v = int((result["tick_volume"] == 0).sum())
+
+    print(f"  保存: {out_file}")
+    print(f"  行数: {len(result)}")
+    print(f"  时间: {first_dt.strftime('%Y-%m-%d %H:%M')} ~ {last_dt.strftime('%Y-%m-%d %H:%M')}")
+    print(f"  跨度: {span_days:.0f} 天 ({span_days/365.25:.1f} 年)")
+    print(f"  收盘价: {result['close'].min():.1f} ~ {result['close'].max():.1f}")
+    print(f"  质量: bad_h={bad_h} bad_l={bad_l} neg={neg} zero_vol={zero_v}")
+
+    # 前后各 3 行
+    print(f"  前3行:")
+    for _, r in result.head(3).iterrows():
+        dt = pd.to_datetime(int(r['time']), unit='s', utc=True).tz_convert(BJ_TZ).strftime('%Y-%m-%d %H:%M')
+        print(f"    {dt}  O={r['open']:.0f} H={r['high']:.0f} L={r['low']:.0f} C={r['close']:.0f} V={int(r['tick_volume'])}")
+    print(f"  后3行:")
+    for _, r in result.tail(3).iterrows():
+        dt = pd.to_datetime(int(r['time']), unit='s', utc=True).tz_convert(BJ_TZ).strftime('%Y-%m-%d %H:%M')
+        print(f"    {dt}  O={r['open']:.0f} H={r['high']:.0f} L={r['low']:.0f} C={r['close']:.0f} V={int(r['tick_volume'])}")
+
+    time.sleep(0.5)
+
+print("\n" + "=" * 80)
+print("全部完成！")
+print("=" * 80)

test_tqsdk_limits.py

@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+"""探索 tqsdk get_kline_serial 的周期和 data_length 上限。"""
+import json
+from pathlib import Path
+from tqsdk import TqApi, TqAuth, TqSim
+
+settings = json.loads(Path(r"D:\cl\AlphaMaster\web_settings.json").read_text(encoding="utf-8"))
+user = settings.get("tqsdk_user", "")
+pwd = settings.get("tqsdk_password", "")
+
+api = TqApi(TqSim(), auth=TqAuth(user, pwd), disable_print=True)
+
+symbol = "KQ.m@SHFE.ru"
+
+# 测试各种周期
+test_durations = [
+    (60, "1m"),
+    (180, "3m"),
+    (300, "5m"),
+    (900, "15m"),
+    (1800, "30m"),
+    (3600, "1h"),
+    (7200, "2h"),
+    (14400, "4h"),
+    (86400, "1d"),
+    (604800, "1w"),
+]
+
+print("=" * 70)
+print("tqsdk 周期 & data_length 上限测试")
+print("=" * 70)
+
+for dur, name in test_durations:
+    # 先试 8000
+    try:
+        df = api.get_kline_serial(symbol, dur, data_length=8000, adj_type="F")
+        got_8000 = len(df)
+    except Exception as e:
+        got_8000 = f"ERROR: {e}"
+        df = None
+
+    # 再试 50000
+    try:
+        df2 = api.get_kline_serial(symbol, dur, data_length=50000, adj_type="F")
+        got_50000 = len(df2)
+    except Exception as e:
+        got_50000 = f"ERROR: {e}"
+
+    # 再试 100000
+    try:
+        df3 = api.get_kline_serial(symbol, dur, data_length=100000, adj_type="F")
+        got_100000 = len(df3)
+    except Exception as e:
+        got_100000 = f"ERROR: {e}"
+
+    print(f"{name:>4s} (dur={dur:>6d}):  len_8000={got_8000}  len_50000={got_50000}  len_100000={got_100000}")
+
+api.close()
+print("done")

verify_data_integrity.py

@@ -0,0 +1,215 @@
+# -*- coding: utf-8 -*-
+"""数据校验脚本：两个维度校验已保存的 parquet 文件。
+
+维度1: 重新拉取 6 个抽样品种的 D1+M15，和已保存文件做逐 bar 对比
+维度2: 对全部 600 个文件做深度一致性检查（时间连续性、价格合理性、成交量合理性）
+"""
+import json
+import time
+from pathlib import Path
+
+import pandas as pd
+import numpy as np
+from datetime import timezone, timedelta
+from tqsdk import TqApi, TqAuth, TqSim
+
+BJ_TZ = timezone(timedelta(hours=8))
+DATA_DIR = Path(r"D:\国内期货K线数据")
+
+settings = json.loads(Path(r"D:\cl\AlphaMaster\web_settings.json").read_text(encoding="utf-8"))
+user = settings.get("tqsdk_user", "")
+pwd = settings.get("tqsdk_password", "")
+if not user or not pwd:
+    user, pwd = "七斗居士", "ghhkphs8"
+
+# ── 维度1: 抽样重新拉取对比 ──
+SAMPLE_SYMBOLS = [
+    ("橡胶主连", "KQ.m@SHFE.ru"),
+    ("沪铜主连", "KQ.m@SHFE.cu"),
+    ("铁矿石主连", "KQ.m@DCE.i"),
+    ("螺纹钢主连", "KQ.m@SHFE.rb"),
+    ("沪深300主连", "KQ.m@CFFEX.IF"),
+    ("碳酸锂主连", "KQ.m@GFEX.lc"),
+]
+
+SAMPLE_TFS = [
+    (900, "M15"),
+    (86400, "D1"),
+]
+
+def fetch_kline(tqsdk_sym, dur, length=10000):
+    api = None
+    try:
+        api = TqApi(TqSim(), auth=TqAuth(user, pwd), disable_print=True)
+        df = api.get_kline_serial(tqsdk_sym, dur, data_length=length, adj_type="F")
+    finally:
+        if api:
+            try: api.close()
+            except: pass
+    if df is None or df.empty:
+        return None
+    df = df[df["datetime"] != 0].copy()
+    df = df[df["volume"] > 0].copy()
+    if len(df) == 0:
+        return None
+    result = pd.DataFrame()
+    result["time"] = (df["datetime"].astype("int64") // 1_000_000_000).astype("int64")
+    result["open"] = df["open"].astype("float64")
+    result["high"] = df["high"].astype("float64")
+    result["low"] = df["low"].astype("float64")
+    result["close"] = df["close"].astype("float64")
+    result["tick_volume"] = df["volume"].astype("int64")
+    result = result.sort_values("time").reset_index(drop=True)
+    result = result.drop_duplicates(subset=["time"], keep="last").reset_index(drop=True)
+    return result
+
+print("=" * 100)
+print("维度1: 抽样重新拉取 vs 已保存文件 逐 bar 对比")
+print("=" * 100)
+
+mismatch_count = 0
+total_compared = 0
+
+for sym_name, tqsdk_sym in SAMPLE_SYMBOLS:
+    for dur, tf_name in SAMPLE_TFS:
+        print(f"\n[{sym_name} {tf_name}] 重新拉取对比中...")
+        fresh = fetch_kline(tqsdk_sym, dur)
+        if fresh is None:
+            print(f"  [ERROR] 重新拉取失败")
+            continue
+
+        saved_path = DATA_DIR / f"{sym_name}_{tf_name}.parquet"
+        saved = pd.read_parquet(saved_path)
+
+        # 找共同的时间戳
+        fresh_times = set(fresh["time"].values)
+        saved_times = set(saved["time"].values)
+        common_times = sorted(fresh_times & saved_times)
+
+        total_compared += 1
+        print(f"  重新拉取: {len(fresh)} rows, 已保存: {len(saved)} rows, 共同: {len(common_times)} rows")
+
+        if len(common_times) == 0:
+            print(f"  [ERROR] 无共同时间戳!")
+            mismatch_count += 1
+            continue
+
+        # 逐 bar 对比
+        fresh_idx = fresh.set_index("time")
+        saved_idx = saved.set_index("time")
+
+        mismatches = []
+        for ts in common_times:
+            f_row = fresh_idx.loc[ts]
+            s_row = saved_idx.loc[ts]
+            for col in ["open", "high", "low", "close", "tick_volume"]:
+                fv = f_row[col]
+                sv = s_row[col]
+                if col == "tick_volume":
+                    if int(fv) != int(sv):
+                        mismatches.append((ts, col, fv, sv))
+                else:
+                    if abs(float(fv) - float(sv)) > 1e-6:
+                        mismatches.append((ts, col, fv, sv))
+
+        if len(mismatches) == 0:
+            print(f"  ✅ 完全一致! {len(common_times)} bars 逐 bar 对比通过")
+        else:
+            mismatch_count += 1
+            print(f"  ❌ 发现 {len(mismatches)} 处不一致:")
+            for ts, col, fv, sv in mismatches[:10]:
+                dt = pd.to_datetime(int(ts), unit="s", utc=True).tz_convert(BJ_TZ).strftime("%Y-%m-%d %H:%M")
+                print(f"    {dt}  {col}: fresh={fv} vs saved={sv}")
+
+        # 检查已保存文件是否比重新拉取多了数据（不应该）
+        extra_saved = sorted(saved_times - fresh_times)
+        if extra_saved:
+            print(f"  ⚠️ 已保存文件多出 {len(extra_saved)} 条 (旧数据残留?)")
+            for ts in extra_saved[:3]:
+                dt = pd.to_datetime(int(ts), unit="s", utc=True).tz_convert(BJ_TZ).strftime("%Y-%m-%d %H:%M")
+                print(f"    {dt}")
+
+        time.sleep(0.5)
+
+print(f"\n维度1 结果: 对比 {total_compared} 个文件, 不一致 {mismatch_count} 个")
+
+# ── 维度2: 全部文件深度一致性检查 ──
+print("\n" + "=" * 100)
+print("维度2: 全部 600 个文件深度一致性检查")
+print("=" * 100)
+
+files = sorted(DATA_DIR.glob("*.parquet"))
+issue_count = 0
+issue_list = []
+
+for f in files:
+    name = f.stem
+    df = pd.read_parquet(f)
+    rows = len(df)
+    issues = []
+
+    # 1. 时间唯一性（不应有重复时间戳）
+    dup_times = df["time"].duplicated().sum()
+    if dup_times > 0:
+        issues.append(f"重复时间戳 {dup_times} 条")
+
+    # 2. 时间升序
+    if not df["time"].is_monotonic_increasing:
+        issues.append("时间未升序排列")
+
+    # 3. OHLC 逻辑
+    o, h, l, c = df["open"], df["high"], df["low"], df["close"]
+    bad_h = ((h < o) | (h < c)).sum()
+    bad_l = ((l > o) | (l > c)).sum()
+    if bad_h > 0:
+        issues.append(f"high<max(O,C) {bad_h} 条")
+    if bad_l > 0:
+        issues.append(f"low>min(O,C) {bad_l} 条")
+
+    # 4. 负价格
+    neg = (df[["open", "high", "low", "close"]] < 0).any(axis=1).sum()
+    if neg > 0:
+        issues.append(f"负价格 {neg} 条")
+
+    # 5. 零成交量
+    zero_v = (df["tick_volume"] == 0).sum()
+    if zero_v > 0:
+        issues.append(f"零成交量 {zero_v} 条 ({zero_v/rows*100:.1f}%)")
+
+    # 6. 时间间隔异常（相邻 bar 时间差应该等于周期）
+    # 只检查 D1（日线最容易判断）
+    if name.endswith("_D1"):
+        diffs = df["time"].diff().dropna()
+        # 日线应该大部分是 86400 秒，周末跳过
+        abnormal_diff = ((diffs != 86400) & (diffs != 172800) & (diffs != 259200) & (diffs != 345600) & (diffs != 604800)).sum()
+        if abnormal_diff > 0:
+            issues.append(f"D1时间间隔异常 {abnormal_diff} 条")
+
+    # 7. 价格跳变检查（相邻 close 涨跌>30%）
+    if rows > 1:
+        rets = np.abs(np.diff(df["close"].values) / df["close"].values[:-1])
+        huge = np.sum(rets > 0.3)
+        if huge > 0:
+            issues.append(f"涨跌>30% {huge} 条")
+
+    # 8. 检查列完整性
+    expected_cols = {"time", "open", "high", "low", "close", "tick_volume"}
+    actual_cols = set(df.columns)
+    if actual_cols != expected_cols:
+        issues.append(f"列不匹配: {actual_cols}")
+
+    if issues:
+        issue_count += 1
+        issue_list.append((name, issues))
+        print(f"  ❌ {name}: {'; '.join(issues)}")
+
+if issue_count == 0:
+    print("\n✅ 全部 600 个文件通过深度一致性检查！")
+else:
+    print(f"\n❌ {issue_count} 个文件有问题:")
+    for name, issues in issue_list:
+        print(f"  {name}: {'; '.join(issues)}")
+
+print("\n" + "=" * 100)
+print(f"校验完成: 维度1 对比 {total_compared} 文件 (不一致 {mismatch_count}), 维度2 检查 {len(files)} 文件 (问题 {issue_count})")
+print("=" * 100)

verify_rubber.py

@@ -0,0 +1,15 @@
+# -*- coding: utf-8 -*-
+"""验证更新后的橡胶主连数据。"""
+import pandas as pd
+from datetime import timezone, timedelta
+
+BJ_TZ = timezone(timedelta(hours=8))
+
+for tf in ['M15', 'M30', 'H1', 'D1']:
+    df = pd.read_parquet(rf'D:\国内期货K线数据\橡胶主连_{tf}.parquet')
+    first = pd.to_datetime(int(df['time'].iloc[0]), unit='s', utc=True).tz_convert(BJ_TZ).strftime('%Y-%m-%d %H:%M')
+    last = pd.to_datetime(int(df['time'].iloc[-1]), unit='s', utc=True).tz_convert(BJ_TZ).strftime('%Y-%m-%d %H:%M')
+    o, h, l, c = df['open'], df['high'], df['low'], df['close']
+    bad_h = ((h < o) | (h < c)).sum()
+    bad_l = ((l > o) | (l > c)).sum()
+    print(f'{tf}: rows={len(df)}, first={first}, last={last}, bad_h={bad_h}, bad_l={bad_l}, neg_price={(c<0).sum()}, zero_vol={(df["tick_volume"]==0).sum()}')