sanguo_vnpy_v2/tests/factor/conftest.py

"""Test configuration for factor module tests."""
import sys
import os

# Add vnpy source to path
_VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vnpy_v4.4.0"))
if _VNPY_SRC not in sys.path:
    sys.path.insert(0, _VNPY_SRC)


# ==================== 合成财务静态域(财务因子批测试共用) ====================
# 3 只股 × 16 报告期(2020Q1~2023Q4,2020 为 SUE 滚动窗预热),NOTICE_DATE 错位:
#   Q1→当年4-28 / H1→当年8-29 / Q3→当年10-27 / 年报→次年4-25
# 数值全部手算可验证(断言用),公式见各 build 函数内注释。
from datetime import date

import polars as pl

REPORT_DATES = [
    f"{y}-{m}" for y in (2020, 2021, 2022, 2023) for m in ("03-31", "06-30", "09-30", "12-31")
]

# 单季 REV/NP 序列(累计=年内前缀和): 报告期索引 i=0..15(2020Q1..2023Q4)
# 2020 为 SUE 滚动窗预热史;2021+ 的断言数值由 2021 段起算
REV_Q = [90, 130, 140, 150] + [100, 120, 150, 160, 110, 140, 150, 180, 130, 140, 170, 180]
NP_Q = [9, 13, 14, 15] + [10, 12, 15, 16, 11, 14, 15, 18, 13, 14, 17, 18]

# 资产负债表(时点存量,随报告期线性演化;2021Q1 起算,idx=i-4)
TA_OF = lambda i: 1000 + 50 * max(i - 4, 0)
EQ_OF = lambda i: 500 + 20 * max(i - 4, 0)
AR_OF = lambda i: 100 + 10 * max(i - 4, 0)
SC_OF = lambda i: 100 if i < 12 else 110  # 2023 起股本扩张 10%(NSI=0.1)

SYN_STOCKS = {
    "600000.SH": {"vt": "600000.SSE", "scale": 1.0,
                  "drop_periods": [], "null_notice_periods": []},
    "000001.SZ": {"vt": "000001.SZSE", "scale": 1.0,
                  "drop_periods": ["2022-03-31"],          # 缺 2022Q1 → 单季差分/TTM 链 NaN
                  "null_notice_periods": ["2022-09-30"]},  # NOTICE_DATE 缺失 → 该报告期跳过
    "300001.SZ": {"vt": "300001.SZSE", "scale": 2.0,
                  "drop_periods": [], "null_notice_periods": []},
}


def _notice_date(report_date: str) -> str:
    """A 股典型披露节奏(年报次年 4-25 / 一季报 4-28 / 中报 8-29 / 三季报 10-27)."""
    y, m = int(report_date[:4]), int(report_date[5:7])
    if m == 3:
        return f"{y}-04-28"
    if m == 6:
        return f"{y}-08-29"
    if m == 9:
        return f"{y}-10-27"
    return f"{y + 1}-04-25"


def _cum_in_year(q_values: list[float], i: int) -> float:
    """报告期 i 的年内累计值 = 当年前几季单季之和."""
    year_start = (i // 4) * 4
    return float(sum(q_values[year_start:i + 1]))


def build_synthetic_static(root: str) -> str:
    """写合成静态域 parquet 树(data_dir),返回 static 根目录路径.

    目录结构与 NAS 一致: static/{income,balance,cashflow}/{code}.{SH|SZ}_{table}.parquet
    forecast 按报告期全市场文件: static/forecast/{YYYYMMDD}_forecast.parquet
    """
    static_dir = os.path.join(str(root), "static")
    for table in ("income", "balance", "cashflow"):
        os.makedirs(os.path.join(static_dir, table), exist_ok=True)

    for file_code, spec in SYN_STOCKS.items():
        s, drop, null_notice = spec["scale"], set(spec["drop_periods"]), set(spec["null_notice_periods"])
        income_rows, balance_rows, cashflow_rows = [], [], []
        for i, rd in enumerate(REPORT_DATES):
            if rd in drop:
                continue
            notice = None if rd in null_notice else _notice_date(rd)
            rev_c, np_c = s * _cum_in_year(REV_Q, i), s * _cum_in_year(NP_Q, i)
            common = {
                "REPORT_DATE": f"{rd} 00:00:00",
                "NOTICE_DATE": (f"{notice} 00:00:00" if notice else None),
                "UPDATE_DATE": f"{notice} 00:00:00" if notice else None,
            }
            income_rows.append({**common,
                "TOTAL_OPERATE_INCOME": rev_c, "OPERATE_COST": 0.6 * rev_c,
                "PARENT_NETPROFIT": np_c, "DEDUCT_PARENT_NETPROFIT": 0.9 * np_c,
                "TOTAL_PROFIT": 1.1 * np_c, "INVEST_INCOME": 0.05 * np_c,
                "FAIRVALUE_CHANGE_INCOME": 0.01 * np_c,
                "ASSET_IMPAIRMENT_LOSS": 0.02 * np_c, "CREDIT_IMPAIRMENT_LOSS": 0.01 * np_c})
            # IBD 只给 SHORT_LOAN 一列(其余组件列缺失,测 schema 缺列容错)
            balance_rows.append({**common,
                "TOTAL_ASSETS": s * TA_OF(i), "TOTAL_PARENT_EQUITY": s * EQ_OF(i),
                "ACCOUNTS_RECE": s * AR_OF(i), "OTHER_RECE": s * (5 + max(i - 4, 0)),
                "GOODWILL": 50.0, "SHARE_CAPITAL": float(SC_OF(i)),
                "SHORT_LOAN": s * (100 + max(i - 4, 0))})
            cashflow_rows.append({**common,
                "NETCASH_OPERATE": 1.2 * np_c, "SALES_SERVICES": 1.05 * rev_c,
                "ACCEPT_INVEST_CASH": 0.1 * rev_c})

        for table, rows in (("income", income_rows), ("balance", balance_rows), ("cashflow", cashflow_rows)):
            pl.DataFrame(rows).write_parquet(
                os.path.join(static_dir, table, f"{file_code}_{table}.parquet"))

    # forecast: 按报告期全市场文件(中文列,归母净利润行优先 + 无净利润行 fallback)
    os.makedirs(os.path.join(static_dir, "forecast"), exist_ok=True)
    pl.DataFrame([
        {"股票代码": "600000", "预测指标": "归属于上市公司股东的净利润",
         "业绩变动幅度": 56.79, "预告类型": "预增", "公告日期": date(2023, 7, 15)},
        {"股票代码": "000001", "预测指标": "归属于上市公司股东的净利润",
         "业绩变动幅度": -30.0, "预告类型": "预减", "公告日期": date(2023, 7, 20)},
        {"股票代码": "300001", "预测指标": "营业收入",
         "业绩变动幅度": 5.0, "预告类型": "略增", "公告日期": date(2023, 7, 10)},
    ]).write_parquet(os.path.join(static_dir, "forecast", "20230630_forecast.parquet"))
    pl.DataFrame([
        {"股票代码": "600000", "预测指标": "净利润",
         "业绩变动幅度": 100.0, "预告类型": "扭亏", "公告日期": date(2023, 10, 15)},
    ]).write_parquet(os.path.join(static_dir, "forecast", "20230930_forecast.parquet"))
    return static_dir


import pytest


@pytest.fixture(scope="session")
def synthetic_static(tmp_path_factory) -> str:
    """session 级合成静态域根目录(test_fundamental_* 共用)."""
    return build_synthetic_static(tmp_path_factory.mktemp("fund_static"))


@pytest.fixture(scope="session")
def np_q_series() -> list[float]:
    """合成归母净利单季序列(SUE 期望值独立重算用)."""
    return list(NP_Q)