添加回测脚本
This commit is contained in:
zlt
parent
40fcf160a3
commit
00a1a20aeb
|
|
@ -56,6 +56,9 @@ SELL_ACTIONS = {"EXIT", "TRIM"} # 真正减仓的落地
|
|||
# 决策系统驱动的判据: 卖出 + 来源是盘中/风控信号 (hard_numbers.source 或 reason 里含这些词)
|
||||
SIGNAL_SOURCES = {"intraday", "risk_sell"}
|
||||
SIGNAL_WORDS = ("决策系统", "风控", "SELL", "转弱", "派发")
|
||||
# 卖出主导信号(hard_numbers.dominant_signal): "take_profit" 这一档是止盈/动量兑现, 其余(转弱、
|
||||
# 派发、大额流出等)归"风控止损"。你担心的"今天买明天卖"正是止盈/动量这拨, 单拎出来判它。
|
||||
TAKE_PROFIT_DOM = "take_profit"
|
||||
# 近似"硬止损"(护栏放行的一档): 入场到卖出这段已经深亏, 视为真出事, 不该被护栏拦
|
||||
HARD_RISK_DROP = -0.08
|
||||
|
||||
|
|
@ -108,7 +111,8 @@ def read_ledger(since_dt: datetime) -> list:
|
|||
|
||||
|
||||
def classify(row: dict) -> dict:
|
||||
"""给账本一行贴标签: side(buy/sell/other) 与 signal_driven(是否决策系统驱动的卖出)。"""
|
||||
"""给账本一行贴标签: side(buy/sell/other)、signal_driven(是否决策系统驱动的卖出)、
|
||||
以及 cohort(止盈动量 / 风控止损, 按 hard_numbers.dominant_signal 分, 只对决策系统卖出有值)。"""
|
||||
act = str(row.get("action") or "").upper()
|
||||
verdict = str(row.get("verdict") or "").upper()
|
||||
acted = verdict == "PASS" # 只认真正放行落地的决策
|
||||
|
|
@ -118,11 +122,17 @@ def classify(row: dict) -> dict:
|
|||
side = "sell"
|
||||
else:
|
||||
side = "other"
|
||||
src = str((row.get("hn") or {}).get("source") or "").lower()
|
||||
hn = row.get("hn") or {}
|
||||
src = str(hn.get("source") or "").lower()
|
||||
reason = str(row.get("reason") or "")
|
||||
signal_driven = (side == "sell") and (
|
||||
src in SIGNAL_SOURCES or any(w in reason for w in SIGNAL_WORDS))
|
||||
return {"side": side, "signal_driven": signal_driven}
|
||||
dom = str(hn.get("dominant_signal") or "").strip().lower()
|
||||
cohort = None
|
||||
if signal_driven:
|
||||
cohort = "止盈动量" if dom == TAKE_PROFIT_DOM else "风控止损"
|
||||
return {"side": side, "signal_driven": signal_driven,
|
||||
"dominant_signal": dom, "cohort": cohort}
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ 历史收盘价源 (gp_day_data @ 18.199)
|
||||
|
|
@ -231,6 +241,7 @@ def pair_round_trips(rows_by_code: dict) -> list:
|
|||
"buy_price": _f(b["price_at"]), "sell_price": _f(s["price_at"]),
|
||||
"hold_days": hold_days, "gross_ret": gross,
|
||||
"signal_driven": tag["signal_driven"], "sell_reason": s.get("reason"),
|
||||
"cohort": tag.get("cohort"),
|
||||
})
|
||||
pending_buy = None
|
||||
return trips
|
||||
|
|
@ -238,13 +249,18 @@ def pair_round_trips(rows_by_code: dict) -> list:
|
|||
|
||||
# ------------------------------------------------------------------ 各段输出
|
||||
def section_inventory(rows, trips):
|
||||
buys = sum(1 for r in rows if classify(r)["side"] == "buy")
|
||||
sells = sum(1 for r in rows if classify(r)["side"] == "sell")
|
||||
sig_sells = sum(1 for r in rows if classify(r)["signal_driven"])
|
||||
tags = [classify(r) for r in rows]
|
||||
buys = sum(1 for t in tags if t["side"] == "buy")
|
||||
sells = sum(1 for t in tags if t["side"] == "sell")
|
||||
sig_sells = sum(1 for t in tags if t["signal_driven"])
|
||||
tp_sells = sum(1 for t in tags if t["cohort"] == "止盈动量")
|
||||
rk_sells = sum(1 for t in tags if t["cohort"] == "风控止损")
|
||||
quick = [t for t in trips if t["signal_driven"] and t["hold_days"] is not None
|
||||
and t["hold_days"] <= max(GUARD_DAYS)]
|
||||
print("\n① 样本盘点")
|
||||
print(f" 账本落地买入 {buys} 笔 · 落地卖出 {sells} 笔 · 其中决策系统驱动的卖出 {sig_sells} 笔")
|
||||
print(f" └ 再拆两拨(按 hard_numbers.dominant_signal): 止盈/动量 {tp_sells} 笔 · "
|
||||
f"风控止损 {rk_sells} 笔")
|
||||
print(f" 配成来回 {len(trips)} 组 · 决策系统驱动的快速来回(≤{max(GUARD_DAYS)}交易日) {len(quick)} 组")
|
||||
return quick
|
||||
|
||||
|
|
@ -266,31 +282,50 @@ def section_roundtrips(trips):
|
|||
f"| {str(t['sell_reason'] or '')[:40]}")
|
||||
|
||||
|
||||
def section_forward(rows):
|
||||
"""③ 前向收益: 卖完之后股价怎么走 —— 是不是瞎折腾的正面回答。"""
|
||||
print("\n③ 决策系统卖出之后的前向收益 (正=卖飞/疑瞎折腾, 负=避坑/该卖)")
|
||||
sig_sells = [r for r in rows if classify(r)["signal_driven"]]
|
||||
def _forward_stats(sells):
|
||||
"""给一批卖出行, 算各前向 horizon 的样本收益序列 {h: [卖后涨跌...]}。"""
|
||||
got = {h: [] for h in FWD_HORIZONS}
|
||||
for r in sig_sells:
|
||||
for r in sells:
|
||||
st = r["decided_at"]
|
||||
st = st if isinstance(st, datetime) else datetime.fromisoformat(str(st))
|
||||
fr = forward_returns(r["ts_code"], st, _f(r["price_at"]))
|
||||
for h in FWD_HORIZONS:
|
||||
if fr[h] is not None:
|
||||
got[h].append(fr[h])
|
||||
return got
|
||||
|
||||
|
||||
def _print_forward(label, got):
|
||||
"""打印一拨卖出的前向收益。至少有一个 horizon 有样本才打印, 返回是否打了。"""
|
||||
if not any(got.values()):
|
||||
print(" (前向价源未接通 fetch_daily_closes, 本段跳过 —— 接上后重跑即出)")
|
||||
return
|
||||
return False
|
||||
print(f" 【{label}】")
|
||||
for h in FWD_HORIZONS:
|
||||
v = got[h]
|
||||
if len(v) < MIN_SAMPLE:
|
||||
print(f" T+{h}: 样本 {len(v)} (<{MIN_SAMPLE}), 只报数")
|
||||
print(f" T+{h}: 样本 {len(v)} (<{MIN_SAMPLE}), 只报数")
|
||||
continue
|
||||
avg = sum(v) / len(v)
|
||||
flew = sum(1 for x in v if x > 0.01) / len(v) # 卖完还涨超1%算卖飞
|
||||
print(f" T+{h}: 样本 {len(v)} · 卖后平均 {_fmt_pct(avg)} · 卖飞比例 {flew:.0%}")
|
||||
print(f" T+{h}: 样本 {len(v)} · 卖后平均 {_fmt_pct(avg)} · 卖飞比例 {flew:.0%}")
|
||||
return True
|
||||
|
||||
|
||||
def section_forward(rows):
|
||||
"""③ 前向收益: 卖完之后股价怎么走 —— 是不是瞎折腾的正面回答。
|
||||
拆三拨: 全部决策系统卖出 / 止盈动量 / 风控止损, 分开看到底谁在瞎折腾。"""
|
||||
print("\n③ 决策系统卖出之后的前向收益 (正=卖飞/疑瞎折腾, 负=避坑/该卖)")
|
||||
tagged = [(r, classify(r)) for r in rows]
|
||||
sig_sells = [r for r, t in tagged if t["signal_driven"]]
|
||||
tp_sells = [r for r, t in tagged if t["cohort"] == "止盈动量"]
|
||||
rk_sells = [r for r, t in tagged if t["cohort"] == "风控止损"]
|
||||
if not _print_forward("全部决策系统卖出", _forward_stats(sig_sells)):
|
||||
print(" (前向价源未接通 fetch_daily_closes, 本段跳过 —— 接上后重跑即出)")
|
||||
return
|
||||
_print_forward("止盈/动量 (你最担心的今买明卖这拨)", _forward_stats(tp_sells))
|
||||
_print_forward("风控止损", _forward_stats(rk_sells))
|
||||
print(" 读法: 卖后平均为正、卖飞比例高 → 这批卖出多在砍还会涨的票(瞎折腾); "
|
||||
"为负 → 多在避坑(该卖)。")
|
||||
"为负 → 多在避坑(该卖)。重点看「止盈/动量」这拨是不是正的 —— 是, 才印证你的担心。")
|
||||
|
||||
|
||||
def section_cost(trips):
|
||||
|
|
@ -314,23 +349,14 @@ def _rt_cost_rate():
|
|||
return COMMISSION_RATE * 2 + STAMP_RATE + TRANSFER_RATE * 2 + SLIPPAGE_BPS / 10000.0 * 2
|
||||
|
||||
|
||||
def section_counterfactual(trips):
|
||||
"""⑤ 反事实: 最小持有期护栏 N 扫一遍。需要前向价源, 缺则跳过。"""
|
||||
print("\n⑤ 反事实: 加「最小持有期护栏」后净收益怎么变 (正=护栏有用)")
|
||||
affected0 = [t for t in trips if t["signal_driven"]]
|
||||
# 探一下前向价源是否可用
|
||||
probe = None
|
||||
for t in affected0:
|
||||
seq = _fwd_closes_after(t["ts_code"], t["sell_at"], n_max=GUARD_HOLD_TO)
|
||||
if seq:
|
||||
probe = True
|
||||
break
|
||||
if not probe:
|
||||
print(" (前向价源未接通, 本段跳过 —— 接上 fetch_daily_closes 后重跑即出)")
|
||||
return
|
||||
def _guard_sweep(label, affected):
|
||||
"""给一拨受影响的来回, 按 N 扫最小持有期护栏, 逐档打印净收益变化。
|
||||
返回 True 表示至少有一档出了结论(样本够), 供上层判空。"""
|
||||
print(f" 【{label}】")
|
||||
printed = False
|
||||
for N in GUARD_DAYS:
|
||||
deltas = []
|
||||
for t in affected0:
|
||||
for t in affected:
|
||||
if t["hold_days"] is None or t["hold_days"] > N:
|
||||
continue # 护栏只管"刚建仓 N 日内"的卖出
|
||||
if (t["gross_ret"] or 0) <= HARD_RISK_DROP:
|
||||
|
|
@ -347,14 +373,38 @@ def section_counterfactual(trips):
|
|||
counter = held - _rt_cost_rate() / 2
|
||||
deltas.append(counter - actual)
|
||||
if len(deltas) < MIN_SAMPLE:
|
||||
print(f" N={N}: 受影响样本 {len(deltas)} (<{MIN_SAMPLE}), 只报数")
|
||||
print(f" N={N}: 受影响样本 {len(deltas)} (<{MIN_SAMPLE}), 只报数")
|
||||
continue
|
||||
avg = sum(deltas) / len(deltas)
|
||||
helped = sum(1 for d in deltas if d > 0) / len(deltas)
|
||||
print(f" N={N} 交易日护栏: 受影响 {len(deltas)} 笔 · 平均净收益变化 {_fmt_pct(avg)} · "
|
||||
printed = True
|
||||
print(f" N={N} 交易日护栏: 受影响 {len(deltas)} 笔 · 平均净收益变化 {_fmt_pct(avg)} · "
|
||||
f"变好占比 {helped:.0%}")
|
||||
return printed
|
||||
|
||||
|
||||
def section_counterfactual(trips):
|
||||
"""⑤ 反事实: 最小持有期护栏 N 扫一遍。需要前向价源, 缺则跳过。
|
||||
拆两拨: 先看全部决策系统卖出, 再单看「止盈/动量」这拨 —— 护栏本就是冲这拨设计的,
|
||||
真要装, 该由它这拨的账说话, 而不是被风控止损那拨稀释。"""
|
||||
print("\n⑤ 反事实: 加「最小持有期护栏」后净收益怎么变 (正=护栏有用)")
|
||||
affected0 = [t for t in trips if t["signal_driven"]]
|
||||
# 探一下前向价源是否可用
|
||||
probe = None
|
||||
for t in affected0:
|
||||
seq = _fwd_closes_after(t["ts_code"], t["sell_at"], n_max=GUARD_HOLD_TO)
|
||||
if seq:
|
||||
probe = True
|
||||
break
|
||||
if not probe:
|
||||
print(" (前向价源未接通, 本段跳过 —— 接上 fetch_daily_closes 后重跑即出)")
|
||||
return
|
||||
_guard_sweep("全部决策系统卖出", affected0)
|
||||
_guard_sweep("止盈/动量 (护栏本就冲这拨设计)",
|
||||
[t for t in trips if t.get("cohort") == "止盈动量"])
|
||||
print(" 读法: 某个 N 上平均变化持续为正且变好占比过半 → 这套信号是「噪音来回型」, "
|
||||
"护栏该上、N 取那档; 若普遍为负 → 是「大波段型」, 别压, 快卖多数是对的。")
|
||||
"护栏该上、N 取那档; 若普遍为负 → 是「大波段型」, 别压, 快卖多数是对的。"
|
||||
"判护栏该不该上, 以「止盈/动量」这拨的账为准。")
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ main
|
||||
|
|
|
|||
Loading…
Reference in New Issue