From 114a69e99781f264d1cf4f64f4774271d2936b45 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Thu, 23 Jul 2026 12:18:41 +0800 Subject: [PATCH] =?UTF-8?q?fix(data):=20dbbardata=20=E6=97=A5=E7=BA=BF?= =?UTF-8?q?=E5=8F=8C=E8=A1=8C=E6=A0=B9=E6=B2=BB(=E7=BB=9F=E4=B8=80?= =?UTF-8?q?=E7=BA=AF=E6=97=A5=E6=9C=9F+helper,=E7=95=99ROWID=20max?= =?UTF-8?q?=E5=8E=BB=E9=87=8D)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因: dbbardata UNIQUE(symbol,exchange,interval,datetime) 按字符串字面比较, 多写入路径混用 'YYYY-MM-DD' 与 'YYYY-MM-DD 00:00:00' -> 同一交易日双行, INSERT OR REPLACE 不去重 -> 回测交易日翻倍/pivot duplicate/信号异常。 方案A (统一纯日期, 详见 Main Agent 诊断): - 新增 scripts/data_platform/dbbardata_utils.py: normalize_daily_dt(s) 取前 10 字符, None/短串安全 - 4 个日线写入脚本写入前调 helper: - bs_eod.py (sanguo-bs-eod 个股日线 baostock) - migrate_daily_baostock.py (历史迁移) - xt_eod.py (sanguo-xt-eod ETF/基金 xtata) - import_vnpy_daily_fast.py (NAS 日线 parquet 导入, 加防御) - TDD: tests/data_platform/test_dbbardata_utils.py 9 cases 全过 - 回归: tests/data_platform + tests/portfolio 199 passed 12 skipped peewee DateTimeField formats 含 '%Y-%m-%d' (阶段0 VPS 实测确认), 读纯日期不崩, 方案A 前提成立。 15min 干净, 不动 (分钟必须带时分)。只改日线 interval='d'。 数据层根治, 不在 provider 适配兜底 (用户铁律)。 --- scripts/data_platform/bs_eod.py | 6 ++- scripts/data_platform/dbbardata_utils.py | 28 +++++++++++ .../data_platform/import_vnpy_daily_fast.py | 6 ++- .../data_platform/migrate_daily_baostock.py | 5 +- scripts/data_platform/xt_eod.py | 8 +++- tests/data_platform/test_dbbardata_utils.py | 46 +++++++++++++++++++ 6 files changed, 92 insertions(+), 7 deletions(-) create mode 100644 scripts/data_platform/dbbardata_utils.py create mode 100644 tests/data_platform/test_dbbardata_utils.py diff --git a/scripts/data_platform/bs_eod.py b/scripts/data_platform/bs_eod.py index d61d347..66fcc23 100644 --- a/scripts/data_platform/bs_eod.py +++ b/scripts/data_platform/bs_eod.py @@ -31,6 +31,8 @@ except (AttributeError, ValueError): import baostock as bs import pandas as pd +from dbbardata_utils import normalize_daily_dt + BASE = Path(r"C:\sanguo_vnpy_v2") DB = BASE / "data" / "quant_trading.db" VAL_DIR = BASE / "data" / "valuation_baostock" @@ -121,10 +123,10 @@ def upsert_daily(conn, code, prefix, rows): "turn", "pctChg", "peTTM", "psTTM", "pcfNcfTTM", "pbMRQ"]: df[c] = pd.to_numeric(df[c], errors="coerce") exc = EXC_MAP[prefix] - # OHLCV -> dbbardata('d') + # OHLCV -> dbbardata('d') — datetime 归一纯日期 (dbbardata 双行根治方案A) db = pd.DataFrame({ "symbol": code, "exchange": exc, - "datetime": df["date"].astype(str) + " 00:00:00", + "datetime": df["date"].astype(str).map(normalize_daily_dt), "interval": "d", "volume": df["volume"], "turnover": df["amount"], "open_interest": 0.0, "open_price": df["open"], "high_price": df["high"], diff --git a/scripts/data_platform/dbbardata_utils.py b/scripts/data_platform/dbbardata_utils.py new file mode 100644 index 0000000..e4af8f9 --- /dev/null +++ b/scripts/data_platform/dbbardata_utils.py @@ -0,0 +1,28 @@ +# -*- coding: utf-8 -*- +"""dbbardata 日线 datetime 归一化 helper (双行根治方案A)。 + +背景: + dbbardata (DbBarData) UNIQUE(symbol,exchange,interval,datetime) 按字符串字面比较。 + 历史多写入路径混用 "YYYY-MM-DD" 与 "YYYY-MM-DD 00:00:00" 两种格式, + 同一交易日 INSERT OR REPLACE 不去重 -> 双行, 致回测翻倍/pivot duplicate。 + +归一规则: + 日线 (interval='d') 写入前必调 normalize_daily_dt, 统一为纯日期 'YYYY-MM-DD'。 + 分钟 (interval='15m' 等) 不要调 — 分钟必须带时分。 + +入口: 所有写 dbbardata interval='d' 的脚本 import 并在 datetime 列写入前调用。 +""" +from typing import Optional + + +def normalize_daily_dt(s: Optional[str]) -> Optional[str]: + """日线 datetime 归一为纯日期 'YYYY-MM-DD' (取前 10 字符)。 + + - 纯日期 'YYYY-MM-DD' -> no-op + - 'YYYY-MM-DD 00:00:00' / 'YYYY-MM-DD HH:MM:SS' -> 截断为 'YYYY-MM-DD' + - None -> None (保留语义不崩溃) + - 短串/空串 -> 原样返回(前 10 字符)不崩溃 + """ + if s is None: + return None + return str(s)[:10] diff --git a/scripts/data_platform/import_vnpy_daily_fast.py b/scripts/data_platform/import_vnpy_daily_fast.py index 2a5bd16..d263d22 100644 --- a/scripts/data_platform/import_vnpy_daily_fast.py +++ b/scripts/data_platform/import_vnpy_daily_fast.py @@ -12,6 +12,8 @@ import sys import time from pathlib import Path +from dbbardata_utils import normalize_daily_dt + DB_PATH = os.environ.get('VNPY_DB_PATH', '/tmp/quant_trading_import.db') DAILY_DIR = os.environ.get('DAILY_DIR', '/Volumes/stock/A股数据/日线数据/daily/') @@ -57,8 +59,8 @@ def import_year(conn, year): combined = pd.concat(all_dfs, ignore_index=True) - # Vectorized conversion - combined['datetime'] = combined['date'].astype(str) + # Vectorized conversion — datetime 归一纯日期 (dbbardata 双行根治方案A, 防御未来混入时分) + combined['datetime'] = combined['date'].astype(str).map(normalize_daily_dt) combined['interval'] = 'd' combined['open_interest'] = 0.0 combined = combined.rename(columns={ diff --git a/scripts/data_platform/migrate_daily_baostock.py b/scripts/data_platform/migrate_daily_baostock.py index 48be76a..5f68da3 100644 --- a/scripts/data_platform/migrate_daily_baostock.py +++ b/scripts/data_platform/migrate_daily_baostock.py @@ -13,6 +13,8 @@ from pathlib import Path import pandas as pd +from dbbardata_utils import normalize_daily_dt + DB = Path(r"C:\sanguo_vnpy_v2\data\quant_trading.db") VAL_DIR = Path(r"C:\sanguo_vnpy_v2\data\valuation_baostock") EXC_MAP = {"SH": "SSE", "SZ": "SZSE"} @@ -52,7 +54,8 @@ for chunk in pd.read_sql("SELECT * FROM daily_baostock_full", c, chunksize=20000 odb = pd.DataFrame({ "symbol": chunk["symbol"].values, "exchange": chunk["exchange"].map(EXC_MAP).values, - "datetime": (chunk["date"].astype(str) + " 00:00:00").values, + # datetime 归一纯日期 (dbbardata 双行根治方案A) + "datetime": chunk["date"].astype(str).map(normalize_daily_dt).values, "interval": "d", "volume": chunk["volume"].values, "turnover": chunk["amount"].values, diff --git a/scripts/data_platform/xt_eod.py b/scripts/data_platform/xt_eod.py index c781062..78baded 100644 --- a/scripts/data_platform/xt_eod.py +++ b/scripts/data_platform/xt_eod.py @@ -21,6 +21,8 @@ import time from xtquant import xtdata as xd import pandas as pd +from dbbardata_utils import normalize_daily_dt + DB = r"C:\sanguo_vnpy_v2\data\quant_trading.db" LOOKBACK = int(__import__("os").environ.get("LOOKBACK_DAYS", "30")) T0 = time.time() @@ -88,8 +90,10 @@ def main(): db = pd.DataFrame({ "symbol": sym, "exchange": exc_of(sym), - "datetime": [str(idx)[:4]+"-"+str(idx)[4:6]+"-"+str(idx)[6:8]+" 00:00:00" - for idx in df.index], + # datetime 归一纯日期 (dbbardata 双行根治方案A) + "datetime": [normalize_daily_dt( + f"{str(idx)[:4]}-{str(idx)[4:6]}-{str(idx)[6:8]}") + for idx in df.index], "interval": "d", "volume": (df["volume"].astype(float).values * 100), "turnover": df["amount"].astype(float).values, diff --git a/tests/data_platform/test_dbbardata_utils.py b/tests/data_platform/test_dbbardata_utils.py new file mode 100644 index 0000000..603c305 --- /dev/null +++ b/tests/data_platform/test_dbbardata_utils.py @@ -0,0 +1,46 @@ +# -*- coding: utf-8 -*- +"""TDD for dbbardata daily datetime normalizer (dbbardata 双行根治方案A).""" +import pytest + +from scripts.data_platform.dbbardata_utils import normalize_daily_dt + + +@pytest.mark.parametrize("s,expected", [ + ("2026-07-20", "2026-07-20"), + ("2010-01-04", "2010-01-04"), +]) +def test_pure_date_noop(s, expected): + # 纯日期应原样返回 + assert normalize_daily_dt(s) == expected + + +@pytest.mark.parametrize("s", [ + "2026-07-20 00:00:00", + "2026-07-20 00:00:00.000000", +]) +def test_truncate_midnight(s): + # 带时间(00:00:00 历史迁移路径) 截断为纯日期 + assert normalize_daily_dt(s) == "2026-07-20" + + +@pytest.mark.parametrize("s", [ + "2026-07-20 15:00:00", + "2026-07-20 09:30:01", +]) +def test_truncate_real_time(s): + # 带真实时分(异常情况) 也截断为纯日期 — 日线无时分语义 + assert normalize_daily_dt(s) == "2026-07-20" + + +def test_none_safe(): + # None 不应崩溃 — 返回 None 保留语义 + assert normalize_daily_dt(None) is None + + +def test_empty_string_safe(): + assert normalize_daily_dt("") == "" + + +def test_short_string_safe(): + # 短于 10 字符的异常输入不崩溃,返回本身(取前 10) + assert normalize_daily_dt("2026") == "2026"