"""四路子因子构造。输入日期区间 [start, end]（YYYY-MM-DD），输出
DataFrame[trade_date, stock_code, factor_value]，全部限制在 KG 覆盖池 universe 内。
stock_code 输出形态不限，common.write_factor 统一转前缀式。

建模参数（EVENT_POLARITY / *_HALF_LIFE / *_WINDOW）是**因子决策**，见设计文档
§3.3 与 §8-3，此处取草案默认，待用户确认后调。方向(direction)统一在合成配置里声明，
子因子表只存原始值——但 event 天然带符号，故此处保留极性。
"""
import numpy as np
import pandas as pd

import config
import common
import db

FACTORS = {
    "akg_upside":       "t_factor_akg_upside",
    "akg_heat":         "t_factor_akg_heat",
    "akg_event":        "t_factor_akg_event",
    "akg_transmission": "t_factor_akg_transmission",
}

# ---- 事件极性草案（§3.3，待 §8-3 确认）----
EVENT_POLARITY = {
    "股份回购": 1.0, "重大合同中标": 1.0, "股权激励授予": 0.5,
    "诉讼仲裁": -1.0, "行政处罚": -1.0, "股权质押": -0.5,
    "发行上市": 0.0, "并购交割": 0.0, "other": 0.0,
    # 增减持 / 业绩预告：符号取决于 direction（下 EVENT_DIR）
    "增减持": 0.0, "业绩预告": 0.0,
}
EVENT_DIR = {"预增": 1.0, "预减": -1.0, "增持": 1.0, "减持": -1.0}
EVENT_HALF_LIFE = 10   # 交易日
EVENT_WINDOW = 60      # 交易日（超窗不计）
_EMPTY = pd.DataFrame(columns=["trade_date", "stock_code", "factor_value"])


# ---------------------------------------------------------------- 热度
def build_heat(start, end):
    """热度 = stock_fund_heat_scores 最新批次 score（0~1）。stock_code 已前缀式。"""
    uni = {common.to_prefix(x) for x in common.load_universe()}
    df = db.read_mysql("heat",
        """SELECT s.trade_date, s.stock_code, s.score AS factor_value
           FROM stock_fund_heat_scores s
           JOIN (SELECT trade_date, MAX(batch_no) bn FROM stock_fund_heat_scores
                 WHERE trade_date BETWEEN %s AND %s GROUP BY trade_date) m
             ON m.trade_date = s.trade_date AND m.bn = s.batch_no""",
        (start, end))
    if df.empty:
        return _EMPTY
    df["stock_code"] = df["stock_code"].astype(str).str.strip()
    df["factor_value"] = pd.to_numeric(df["factor_value"], errors="coerce")
    df = df[df["stock_code"].isin(uni)]
    return df[["trade_date", "stock_code", "factor_value"]]


def _read_gp_price(start, end):
    """gp_day_data 现价：代码列名不定（平台实测=symbol，非 ts_code）——照 rsi_14d_etl
    惯例按候选列逐个试，用第一个能查通的。"""
    cands = [config.PRICE_CODE_COL] + [c for c in ("symbol", "ts_code")
                                       if c != config.PRICE_CODE_COL]
    last = None
    for c in cands:
        try:
            df = db.read_mysql("price",
                f"SELECT `timestamp` AS trade_date, `{c}` AS ts_code, close "
                f"FROM gp_day_data WHERE `timestamp` BETWEEN %s AND %s", (start, end))
            print(f"  (upside 现价用 gp_day_data.{c})")
            return df
        except Exception as e:  # noqa: BLE001 —— 列名不对就换下一个候选
            last = e
    raise RuntimeError(f"gp_day_data 代码列都不行（试了 {cands}）: {last!r}")


# ---------------------------------------------------------------- 预期空间
def build_upside(start, end):
    """upside = 一致预期目标价中枢 / 当日现价 − 1（as-of：现价日取 asof<=当日最新一致预期）。
    现价来自平台 gp_day_data（代码列 config.PRICE_CODE_COL，待实机核实）。"""
    uni = common.load_universe()
    cons = db.read_pg(
        "SELECT ts_code, asof_date, target_mid_avg FROM v_factor_consensus WHERE asof_date <= %s",
        (end,))
    cons = cons[cons["ts_code"].isin(uni)].copy()
    if cons.empty:
        return _EMPTY
    price = _read_gp_price(start, end)
    if price.empty:
        return _EMPTY
    price["close"] = pd.to_numeric(price["close"], errors="coerce")
    # 归一到前缀式两边对齐（gp_day_data 代码形态不定 → 都折前缀式后 join）
    price["k"] = price["ts_code"].map(common.to_prefix)
    price = price[(price["close"] > 0)].dropna(subset=["close"])
    cons["k"] = cons["ts_code"].map(common.to_prefix)
    cons = cons[cons["k"].isin(set(price["k"]))]
    if cons.empty:
        return _EMPTY
    # 强制两侧键同分辨率 datetime64[ns]（pandas 2.x 不同来源可能 us/ns 混，
    # merge_asof 会报 incompatible merge keys）
    cons["asof_date"] = pd.to_datetime(cons["asof_date"]).astype("datetime64[ns]")
    price["trade_date"] = pd.to_datetime(price["trade_date"]).astype("datetime64[ns]")
    left = price[["trade_date", "k", "close"]].sort_values("trade_date")
    right = cons[["asof_date", "k", "target_mid_avg"]].sort_values("asof_date")
    m = pd.merge_asof(left, right, left_on="trade_date", right_on="asof_date",
                      by="k", direction="backward")          # 每股取 asof<=当日最新目标价
    m = m.dropna(subset=["target_mid_avg", "close"])
    if m.empty:
        return _EMPTY
    m["factor_value"] = m["target_mid_avg"].astype(float) / m["close"] - 1.0
    m = m.rename(columns={"k": "stock_code"})
    return m[["trade_date", "stock_code", "factor_value"]]


# ---------------------------------------------------------------- 事件
def _polarity(event_type, direction):
    d = (direction or "").strip()
    if d in EVENT_DIR:
        return EVENT_DIR[d]
    return EVENT_POLARITY.get(event_type, 0.0)


def build_event(start, end):
    """事件分 = Σ 近窗口内事件 极性 × 时间衰减（exp(-交易日龄·ln2/半衰期)）。
    ts_code 取文档锚（v_factor_events 已解析）。无事件的股当天不出行（= 缺 → 合成侧填 0）。"""
    uni = common.load_universe()
    look = (pd.Timestamp(start) - pd.Timedelta(days=EVENT_WINDOW * 2)).date()
    ev = db.read_pg(
        "SELECT ts_code, disclosure_date, event_type, direction "
        "FROM v_factor_events WHERE disclosure_date BETWEEN %s AND %s", (look, end))
    ev = ev[ev["ts_code"].isin(uni)].copy()
    if ev.empty:
        return _EMPTY
    ev["event_type"] = ev["event_type"].fillna("")
    ev["direction"] = ev["direction"].fillna("")   # 多数事件无 direction(NULL→NaN)，先填空防 .strip 崩
    ev["pol"] = [_polarity(t, d) for t, d in zip(ev["event_type"], ev["direction"])]
    ev = ev[ev["pol"] != 0.0]
    if ev.empty:
        return _EMPTY
    cal = common.trading_days(start, end)
    if not cal:
        return _EMPTY
    cal = pd.DatetimeIndex(cal)
    decay = np.log(2) / EVENT_HALF_LIFE
    rows = []
    for ts, g in ev.groupby("ts_code"):
        disc = pd.to_datetime(g["disclosure_date"]).values.astype("datetime64[ns]")
        pol = g["pol"].to_numpy(dtype=float)
        for d in cal:
            # 自然日龄 → 交易日龄近似 ×(5/7)（v1 近似，见 README 待优化项）
            age_td = ((d.value - disc.astype("int64")) / 86_400e9) * (5.0 / 7.0)
            mask = (age_td >= 0) & (age_td <= EVENT_WINDOW)
            if not mask.any():
                continue
            val = float((pol[mask] * np.exp(-age_td[mask] * decay)).sum())
            if val != 0.0:
                rows.append((d.date(), ts, val))
    return pd.DataFrame(rows, columns=["trade_date", "stock_code", "factor_value"]) if rows else _EMPTY


# ---------------------------------------------------------------- 传导
def build_transmission(start, end):
    """传导分 = 指向该股所在环节的路径数 ×（1 − 已动比例）；同股同日多候选取最大。"""
    uni = common.load_universe()
    tr = db.read_pg(
        "SELECT scan_date, ts_code, n_paths, moved_ratio "
        "FROM v_factor_transmission WHERE scan_date BETWEEN %s AND %s", (start, end))
    tr = tr[tr["ts_code"].isin(uni)].copy()
    if tr.empty:
        return _EMPTY
    tr["factor_value"] = (tr["n_paths"].astype(float)
                          * (1.0 - pd.to_numeric(tr["moved_ratio"], errors="coerce").fillna(0.0)))
    g = (tr.groupby(["scan_date", "ts_code"])["factor_value"].max().reset_index()
         .rename(columns={"scan_date": "trade_date", "ts_code": "stock_code"}))
    return g[["trade_date", "stock_code", "factor_value"]]


BUILDERS = {
    "akg_upside": build_upside, "akg_heat": build_heat,
    "akg_event": build_event, "akg_transmission": build_transmission,
}