179 lines
8.3 KiB
Python
179 lines
8.3 KiB
Python
"""四路子因子构造。输入日期区间 [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,
|
||
}
|