# sanguo_factor/fundamental_forecast.py
"""forecast 业绩预告事件层: static/forecast 按报告期全市场文件 → 事件流
(自 fundamental_adapter.py 拆出,纯结构重构).

fc_events(公告日 asof 信号)+ fc_pair(F07 预告兑现差配对原料)双出口;
归母净利润行优先,年化预告净利只对净利行生效.
"""
from __future__ import annotations

import os
from datetime import datetime

import polars as pl

from .fundamental_schema import FORECAST_TYPE_SCORE

# ==================== forecast 事件层 ====================

# 预告净利年化系数(按报告期进度;D13 预期 EP): Q1×4 / H1×2 / Q3×4/3 / 年报×1
_ANNUALIZE_FACTOR = {3: 4.0, 6: 2.0, 9: 4.0 / 3.0, 12: 1.0}

# PIT 生效日偏移开关(eff+N 敏感性对照实验,调用时读取). 预告常在收盘后公告,
# 默认 0 = 公告日 D 当天即可用其信息;设 N 时所有预告事件生效日整体推后 N 个
# 自然日(D 日不可见,D+N 起可见),同时作用于 fc_events 与 fc_pair(F07 兑现差
# 锚的预告腿);三表/分红/户数/十大流通股东/vb 域不受此开关影响.
# 跑批侧两种设置方式(取其一):
#   ①容器/进程环境变量: SANGUO_FORECAST_PIT_OFFSET_DAYS=1(docker run -e 即可)
#   ②进程内直设(batch_eval 在调 iter_fundamental_feature_chunks 前):
#     import sanguo_factor.fundamental_forecast as _ff
#     _ff.FORECAST_PIT_OFFSET_DAYS = 1
FORECAST_PIT_OFFSET_DAYS = int(os.environ.get("SANGUO_FORECAST_PIT_OFFSET_DAYS", "0") or 0)


def _load_forecast_events(codes: list[str], data_dir: str) -> tuple[pl.DataFrame, pl.DataFrame]:
    """forecast 按期文件 → (fc_events, fc_pair).

    fc_events: (vt_symbol, eff=公告日期, forecast_type_score, forecast_change_pct,
                forecast_np_annualized) 事件行——一股一公告日多行(按预测指标),
    归母净利润行优先(含"净利润"且不含"扣"),无净利润行 fallback 任意行;
    年化预告净利只对净利行生效(fallback 营业收入行的中值不作净利用)。
    同股多公告日全保留(asof 取最新)。

    fc_pair: (vt_symbol, REPORT_DATE, _fc_mid, eff) —— F07 预告兑现差的配对原料,
    每 (股, 报告期) 取最新公告日的净利行中值(REPORT_DATE 取自文件名)。

    生效日 eff 受模块级 FORECAST_PIT_OFFSET_DAYS 偏移(默认 0=公告日本身),
    两出口(fc_events/fc_pair)同步生效。
    """
    schema = {"vt_symbol": pl.Utf8, "eff": pl.Date, "REPORT_DATE": pl.Date,
              "forecast_type_score": pl.Float64, "forecast_change_pct": pl.Float64,
              "forecast_np_annualized": pl.Float64, "_fc_mid": pl.Float64, "_is_np": pl.Boolean}
    fc_dir = os.path.join(data_dir, "forecast")
    if not os.path.isdir(fc_dir):
        return pl.DataFrame(schema=schema), pl.DataFrame(schema=schema)
    code_set = set(codes)
    frames = []
    for fname in sorted(os.listdir(fc_dir)):
        if not fname.endswith(".parquet"):
            continue
        try:
            f = pl.read_parquet(os.path.join(fc_dir, fname))
        except Exception:
            continue
        if f.height == 0 or not all(c in f.columns for c in
                                    ("股票代码", "预告类型", "公告日期")):
            continue
        try:  # 文件名前 8 位 = 报告期(20230630_forecast.parquet)
            report_date = datetime.strptime(fname[:8], "%Y%m%d").date()
        except ValueError:
            continue
        code = pl.col("股票代码").cast(pl.Utf8).str.strip_chars().str.zfill(6)
        eff = pl.col("公告日期").cast(pl.Date, strict=False)
        if FORECAST_PIT_OFFSET_DAYS:  # eff+N 敏感性开关(见模块级常量注释)
            eff = eff.dt.offset_by(f"{FORECAST_PIT_OFFSET_DAYS}d")
        # 交易所映射: 60→SSE;北交前缀白名单(92/43/82/83)→BJSE;其余→SZSE
        # (互评备注: 北交种类不得落入 SZSE——容器/实盘 universe 按后缀路由)
        is_bj = (code.str.starts_with("92") | code.str.starts_with("43")
                 | code.str.starts_with("82") | code.str.starts_with("83"))
        vt = (pl.when(code.str.starts_with("60")).then(code + pl.lit(".SSE"))
              .when(is_bj).then(code + pl.lit(".BJSE"))
              .otherwise(code + pl.lit(".SZSE")).alias("vt_symbol"))
        if "预测指标" in f.columns:
            ind = pl.col("预测指标").cast(pl.Utf8)
            pref = (ind.str.contains("净利润") & ~ind.str.contains("扣")).cast(pl.Int32)
        else:
            pref = pl.lit(0, pl.Int32)
        pct = (pl.col("业绩变动幅度").cast(pl.Float64, strict=False)
               if "业绩变动幅度" in f.columns else pl.lit(None, pl.Float64))
        mid = (pl.col("预测数值").cast(pl.Float64, strict=False)
               if "预测数值" in f.columns else pl.lit(None, pl.Float64))
        f = f.with_columns(
            vt,
            pref.alias("_pref"),
            pct.alias("_pct"),
            mid.alias("_mid"),
            pl.lit(report_date, dtype=pl.Date).alias("REPORT_DATE"),
            pl.lit(_ANNUALIZE_FACTOR.get(report_date.month), dtype=pl.Float64).alias("_annf"),
            eff.alias("eff"),
            pl.col("预告类型").cast(pl.Utf8).replace_strict(
                FORECAST_TYPE_SCORE, default=None, return_dtype=pl.Float64
            ).alias("_score"),
        ).filter(pl.col("vt_symbol").is_in(code_set) & pl.col("eff").is_not_null())
        if f.height:
            frames.append(f.select(
                ["vt_symbol", "eff", "REPORT_DATE", "_pref", "_score", "_pct", "_mid", "_annf"]))
    if not frames:
        return pl.DataFrame(schema=schema), pl.DataFrame(schema=schema)
    fc = pl.concat(frames).sort(["vt_symbol", "eff", "_pref"])
    # 同 (vt, 公告日, 报告期) 取优先级最高行(_pref 大者排序在后 → last);
    # 年化预告净利 = 净利行中值 × 年化系数(非净利行 fallback → null)
    events = fc.group_by(["vt_symbol", "eff", "REPORT_DATE"]).agg(
        pl.col("_score").last().alias("forecast_type_score"),
        pl.col("_pct").last().alias("forecast_change_pct"),
        pl.col("_pref").last().alias("_is_np"),
        pl.col("_mid").last().alias("_mid"),
        pl.col("_annf").last().alias("_annf"),
    ).with_columns(
        pl.when(pl.col("_is_np") == 1)
        .then(pl.col("_mid") * pl.col("_annf")).otherwise(None)
        .alias("forecast_np_annualized"),
    )
    # F07 配对: 每 (股, 报告期) 最新公告日的净利行中值
    fc_pair = (events.filter(pl.col("_is_np") == 1 & pl.col("_mid").is_not_null())
               .sort(["vt_symbol", "REPORT_DATE", "eff"])
               .group_by(["vt_symbol", "REPORT_DATE"]).agg(
                   pl.col("eff").last().alias("eff"),
                   pl.col("_mid").last().alias("_fc_mid"))
               .select(["vt_symbol", "REPORT_DATE", "_fc_mid", "eff"]))
    # 事件流: 同 (股, 公告日) 多报告期行罕见(同年同日两期预告)——取最新报告期
    # 为当前信号(P0 语义 = 每公告日一行)
    fc_events = (events.sort(["vt_symbol", "eff", "REPORT_DATE"])
                 .group_by(["vt_symbol", "eff"]).last()
                 .select(["vt_symbol", "eff", "forecast_type_score",
                          "forecast_change_pct", "forecast_np_annualized"]))
    return fc_events, fc_pair