diff --git a/backend/app/services/financial_sync.py b/backend/app/services/financial_sync.py index 6655496..6562bf4 100644 --- a/backend/app/services/financial_sync.py +++ b/backend/app/services/financial_sync.py @@ -160,7 +160,8 @@ def _merge_report_history(*frames: pl.DataFrame) -> pl.DataFrame: 语义(区分"覆盖"与"填空"): 每列独立取 announce_date 最新的非空值 — 新同步行有值则覆盖旧值, 新行缺的列(如 fuyao 不提供的字段)由旧行补齐, 实现多数据源并集共存。历史报告期不可变, 合并不会引入过期数据。 - 无 announce_date 的帧按输入顺序, 后写优先(与旧行为 keep="last" 一致)。 + 无 announce_date 的帧按输入顺序, 后写优先(与旧行为 keep="last" 一致); + 公告日为空视为最旧, 不得压过带公告日的行。 """ valid = [ frame @@ -176,7 +177,10 @@ def _merge_report_history(*frames: pl.DataFrame) -> pl.DataFrame: sort_keys = ["symbol", "period_end"] + ( ["announce_date"] if "announce_date" in merged.columns else [] ) - merged = merged.sort(sort_keys, nulls_last=True) + # 公告日为空排在最前: 排到最后会让"公告日未知"的旧行在逐列 last() 时胜出, + # 产出 announce_date 是新公告、数值却是旧值的自相矛盾行。symbol/period_end + # 已在上面过滤掉空值, 不受该参数影响。 + merged = merged.sort(sort_keys, nulls_last=False) value_cols = [c for c in merged.columns if c not in ("symbol", "period_end")] return ( merged.group_by("symbol", "period_end") diff --git a/backend/tests/test_fundamental_factors.py b/backend/tests/test_fundamental_factors.py index c148678..9e4bc8f 100644 --- a/backend/tests/test_fundamental_factors.py +++ b/backend/tests/test_fundamental_factors.py @@ -210,3 +210,41 @@ def test_financial_sync_merges_history(tmp_path: Path): assert row.height == 1 assert row["roe"].item() == 9.5 assert merged2.height == 2 # 修正不增加行数 + + +def test_financial_sync_merge_unknown_announce_date_does_not_win(): + """公告日未知的旧行不得压过带公告日的新行。 + + 多源并存时旧行可能没有 announce_date (provider 不提供 / 上游缺该字段)。 + 这类"公告日未知"的行若排在真实公告日之后, 逐列取最后一个非空值时反而胜出, + 合并结果会出现「announce_date 是新公告、数值仍是旧值」的自相矛盾行, + 点时因子据此在公告日之后放出的是修正前的数。 + """ + from app.services import financial_sync as fs + + revised = pl.DataFrame({ + "symbol": ["600000.SH"], + "period_end": ["2025-12-31"], + "announce_date": ["2026-02-01"], + "roe": [9.5], + }) + # 旧行有 announce_date 列但取值为空 + old_null = pl.DataFrame({ + "symbol": ["600000.SH"], + "period_end": ["2025-12-31"], + "announce_date": [None], + "roe": [9.0], + }) + row = fs._merge_report_history(old_null, revised).to_dicts()[0] + assert row["announce_date"] == "2026-02-01" + assert row["roe"] == 9.5 + + # 旧帧整列缺失 (另一数据源不提供该字段) — 文档承诺"后写优先" + old_missing = pl.DataFrame({ + "symbol": ["600000.SH"], + "period_end": ["2025-12-31"], + "roe": [9.0], + }) + row = fs._merge_report_history(old_missing, revised).to_dicts()[0] + assert row["announce_date"] == "2026-02-01" + assert row["roe"] == 9.5