添加回测脚本
This commit is contained in:
zlt
parent
00a1a20aeb
commit
6aaa4398d6
|
|
@ -0,0 +1,492 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
入场体检: 决策系统的买入, 有多少是"买完就被风控清掉"的坏入场, 该不该在入场端加一道过滤 (只读)
|
||||
==============================================================================
|
||||
承接换手体检 backtest_churn.py 的结论: 决策系统的快速卖出这一批全是风控止损, 而且基本都砍对了
|
||||
(卖完股价还在跌), 加"卖出端最小持有期护栏"只会更差。于是问题被顶到了另一头 —— 不是卖得太快,
|
||||
是买得不对: 系统反复买那种一两天内就触发风控、只能亏着清掉的票。这个脚本把这批"坏入场"拎出来,
|
||||
量三件事:
|
||||
一, 它们买入时的价格情形有没有共性 (是不是买在冲高、买在已经下跌的票上);
|
||||
二, 买入之后股价怎么走 (入场时点本身好不好, 跟全体入场比);
|
||||
三, 若在入场端加一道过滤, 是净赚 (挡掉的多是亏的), 还是误伤好票 (挡掉的里不少是赚的)。
|
||||
|
||||
一句话方法: 同样只读评审账本 pms_action_ledger, 把"一次买紧跟一次卖"配成来回。来回里"卖出是
|
||||
决策系统驱动、且持有不超过 N 个交易日"的那批, 就是坏入场。买入时点的价格情形用 gp_day_data 的
|
||||
当日 OHLC 重建 (它有 open/high/low/close/volume, 见 DATA_MODEL §1.1)。
|
||||
|
||||
五段:
|
||||
① 样本盘点 全部买入多少、配成来回多少、其中"买完就被风控快速清掉"的坏入场多少。
|
||||
② 坏入场画像 坏入场 vs 全体入场, 比买入日的三个价格特征: 追高度、当日涨跌、前5日动量。
|
||||
③ 入场后前向 每笔买入之后 T+1/5/20 相对买价的涨跌, 坏入场 vs 全体, 看入场时点本身好不好。
|
||||
④ 成本账 这批坏入场的来回一共交了多少纯摩擦。
|
||||
⑤ 反事实过滤 扫几条入场过滤规则(追高/逆势/放量阴线): 各挡掉多少来回、其中坏入场多少(召回)、
|
||||
挡掉的平均实现净收益、误伤率、以及组合净效果。挡掉的多是亏的且净效果为正=该过滤
|
||||
有用; 误伤率高(挡掉的里不少是赚的)=别上。
|
||||
|
||||
成本口径与前向价源同 backtest_churn.py (往返≈0.262%; gp_day_data @ 18.199 走 app 的 index 源)。
|
||||
读数纪律: 样本少于 MIN_SAMPLE 的段落只列数、不下结论 —— 不等数据的判分是编故事。
|
||||
|
||||
一处口径说明: 判"坏入场"用的是卖出的 dominant_signal 与来源, 与换手体检同口径; 只认"决策系统
|
||||
驱动的卖出"配成的来回。到价止盈那类计划位平仓不在决策系统卖出信号里(它走执行层的目标价),
|
||||
所以不进这里的坏入场, 也不该进 —— 这段量的就是"被风控快速清掉"的那类入场。
|
||||
|
||||
运行(桥机 factorevaluation, 项目根目录):
|
||||
docker compose run --rm pms-web python scripts/backtest_entry.py --days 120
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from app.db.session import fetch_all, DBUnavailable # noqa: E402 只读单表, 过单表守卫
|
||||
|
||||
# ------------------------------------------------------------------ 可调常量
|
||||
MIN_SAMPLE = 5 # 少于这个数只报数不下结论
|
||||
|
||||
COMMISSION_RATE = 0.00025 # 佣金费率(每边); 万2.5
|
||||
COMMISSION_MIN = 5.0 # 佣金每笔最低(元)
|
||||
STAMP_RATE = 0.0005 # 印花税(仅卖出); 千0.5
|
||||
TRANSFER_RATE = 0.00001 # 过户费(双边); 十万1
|
||||
SLIPPAGE_BPS = 8.0 # 滑点(每边, 基点); 8bp = 0.08%
|
||||
|
||||
# 账本 action 归类 (与 backtest_churn.py 同口径)
|
||||
BUY_ACTIONS = {"OPEN", "FILL", "ADD", "DCA"}
|
||||
SELL_ACTIONS = {"EXIT", "TRIM"}
|
||||
SIGNAL_SOURCES = {"intraday", "risk_sell"}
|
||||
SIGNAL_WORDS = ("决策系统", "风控", "SELL", "转弱", "派发")
|
||||
TAKE_PROFIT_DOM = "take_profit" # 卖出 dominant_signal 是这个 = 止盈离场, 其余 = 风控止损
|
||||
|
||||
# "坏入场"判据: 买入配成的来回, 卖出是决策系统驱动、且持有不超过这么多交易日(自然日近似)
|
||||
QUICK_DAYS = 10
|
||||
FWD_HORIZONS = (1, 5, 20) # 入场后前向看这几个交易日
|
||||
|
||||
# 入场过滤规则的阈值 (⑤ 反事实扫的候选闸门, 都只用买入日及之前的信息, 无未来函数)
|
||||
CHASE_TOP_FRAC = 0.70 # 追高: 买价落在当日振幅的顶部这个比例以上 (1.0=买在最高)
|
||||
DOWNTREND_5D = -0.03 # 逆势: 买入日相对前5个交易日收盘已跌超这个幅度
|
||||
REDVOL_VOL_MULT = 1.5 # 放量阴线: 买入日是阴线且量能≥前5日均量的这个倍数
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ 小工具
|
||||
def _f(v, d=0.0):
|
||||
try:
|
||||
return float(v)
|
||||
except (TypeError, ValueError):
|
||||
return d
|
||||
|
||||
|
||||
def _load_json(s):
|
||||
if not s:
|
||||
return {}
|
||||
try:
|
||||
return json.loads(s) if isinstance(s, str) else dict(s)
|
||||
except (json.JSONDecodeError, TypeError, ValueError):
|
||||
return {}
|
||||
|
||||
|
||||
def _pct(entry, exit_):
|
||||
e, x = _f(entry), _f(exit_)
|
||||
return (x / e - 1.0) if e > 0 and x > 0 else None
|
||||
|
||||
|
||||
def _fmt_pct(x):
|
||||
return f"{x:+.2%}" if x is not None else "—"
|
||||
|
||||
|
||||
def _avg(xs):
|
||||
xs = [x for x in xs if x is not None]
|
||||
return (sum(xs) / len(xs)) if xs else None
|
||||
|
||||
|
||||
def _rt_cost_rate():
|
||||
return COMMISSION_RATE * 2 + STAMP_RATE + TRANSFER_RATE * 2 + SLIPPAGE_BPS / 10000.0 * 2
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ 读账本 (单表, 过守卫)
|
||||
def read_ledger(since_dt: datetime) -> list:
|
||||
"""按时间读评审账本。只读一张表, 满足 153 代理的严格单表访问。"""
|
||||
rows = fetch_all(
|
||||
"SELECT id, ts_code, decided_at, action, arbiter, verdict, price_at, "
|
||||
"hard_numbers_json, reason, ref_id "
|
||||
"FROM pms_action_ledger WHERE decided_at >= :since "
|
||||
"ORDER BY ts_code, decided_at, id",
|
||||
{"since": since_dt.strftime("%Y-%m-%d %H:%M:%S")}, source="proxy")
|
||||
out = []
|
||||
for r in rows:
|
||||
r = dict(r)
|
||||
r["hn"] = _load_json(r.get("hard_numbers_json"))
|
||||
out.append(r)
|
||||
return out
|
||||
|
||||
|
||||
def classify(row: dict) -> dict:
|
||||
"""给账本一行贴标签: 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" # 只认真正放行落地的决策
|
||||
if act in BUY_ACTIONS and acted:
|
||||
side = "buy"
|
||||
elif act in SELL_ACTIONS and acted:
|
||||
side = "sell"
|
||||
else:
|
||||
side = "other"
|
||||
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))
|
||||
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)
|
||||
_BAR_CACHE: dict = {}
|
||||
_ANALYSIS_START = None # main 里设; 让按整个分析区间取足够宽的每股窗口
|
||||
_ANALYSIS_END = None
|
||||
|
||||
|
||||
def _to_prefix(ts_code: str) -> str:
|
||||
"""点式 600000.SH → 前缀式 SH600000 (gp_day_data.symbol 用前缀式, 见 DATA_MODEL 约定)。"""
|
||||
s = str(ts_code or "").strip().upper()
|
||||
if "." in s:
|
||||
num, ex = s.split(".", 1)
|
||||
return ex + num
|
||||
return s
|
||||
|
||||
|
||||
def fetch_daily_bars(ts_code: str, start_dt: datetime, end_dt: datetime) -> dict:
|
||||
"""返回 {date: {open,high,low,close,vol,pre_close,pct}}, 只含有行情的交易日。
|
||||
|
||||
源: gp_day_data (db_gp_cj @ 192.168.18.199, 走 app 的 index 数据源)。按 DATA_MODEL §1.1:
|
||||
symbol 是**前缀式**(这里转一下); open/high/low/close 是 **VARCHAR**(读出转 float);
|
||||
percent、pre_close 是 DECIMAL。用原始价(非前复权), 与账本 price_at 同口径。
|
||||
库不可达或该股无行情返回 {}, ②③⑤ 相关段落自动跳过、不报错。
|
||||
"""
|
||||
sym = _to_prefix(ts_code)
|
||||
try:
|
||||
rows = fetch_all(
|
||||
"SELECT `timestamp` AS d, `open` AS o, `high` AS h, `low` AS l, `close` AS c, "
|
||||
"`volume` AS v, `pre_close` AS pc, `percent` AS pct FROM gp_day_data "
|
||||
"WHERE symbol = :s AND `timestamp` >= :a AND `timestamp` <= :b "
|
||||
"ORDER BY `timestamp`",
|
||||
{"s": sym, "a": start_dt.strftime("%Y-%m-%d 00:00:00"),
|
||||
"b": end_dt.strftime("%Y-%m-%d 23:59:59")}, source="index")
|
||||
except Exception:
|
||||
return {}
|
||||
out = {}
|
||||
for r in rows:
|
||||
d = r.get("d")
|
||||
if d is None:
|
||||
continue
|
||||
dd = d.date() if isinstance(d, datetime) else datetime.fromisoformat(str(d)).date()
|
||||
close = _f(r.get("c"))
|
||||
if close <= 0:
|
||||
continue
|
||||
out[dd] = {"open": _f(r.get("o")), "high": _f(r.get("h")), "low": _f(r.get("l")),
|
||||
"close": close, "vol": _f(r.get("v")),
|
||||
"pre_close": _f(r.get("pc")), "pct": _f(r.get("pct"))}
|
||||
return out
|
||||
|
||||
|
||||
def _bars(ts_code: str, anchor_dt: datetime) -> dict:
|
||||
"""按股缓存**整个分析区间**的日线(带前后余量), 覆盖该股所有买卖点。"""
|
||||
if ts_code not in _BAR_CACHE:
|
||||
lo = (_ANALYSIS_START or anchor_dt) - timedelta(days=30) # 前多留些, 好算前5日动量
|
||||
hi = (_ANALYSIS_END or anchor_dt) + timedelta(days=60)
|
||||
_BAR_CACHE[ts_code] = fetch_daily_bars(ts_code, lo, hi)
|
||||
return _BAR_CACHE[ts_code]
|
||||
|
||||
|
||||
def _closes_after(ts_code: str, dt: datetime) -> list:
|
||||
"""某日之后的交易日收盘序列(升序), 供取前向 T+1/T+5/T+20。"""
|
||||
bars = _bars(ts_code, dt)
|
||||
if not bars:
|
||||
return []
|
||||
d0 = dt.date()
|
||||
after = sorted((d, b["close"]) for d, b in bars.items() if d > d0)
|
||||
return [c for _, c in after]
|
||||
|
||||
|
||||
def forward_returns(ts_code: str, dt: datetime, base_price: float) -> dict:
|
||||
"""dt 之后 T+h 相对 base_price 的涨跌。"""
|
||||
seq = _closes_after(ts_code, dt)
|
||||
return {h: (_pct(base_price, seq[h - 1]) if len(seq) >= h else None) for h in FWD_HORIZONS}
|
||||
|
||||
|
||||
def entry_context(ts_code: str, buy_dt: datetime, buy_price: float) -> dict:
|
||||
"""重建买入时点的价格情形。返回 {chase, day_chg, mom5, red_vol}, 缺数据的项为 None。
|
||||
chase 追高度 = (买价-当日最低)/(当日最高-当日最低), 1.0=买在最高, 越高越追。
|
||||
day_chg 买入日涨跌 = 当日 percent(优先) 或 收盘/前收-1, 正=买在红盘。
|
||||
mom5 前5日动量 = 买入日收盘/前5个交易日收盘-1, 负=买在已经下跌的票上(逆势)。
|
||||
red_vol 放量阴线 = 买入日收盘<开盘 且 量能≥前5日均量×倍数 (True/False/None)。
|
||||
"""
|
||||
bars = _bars(ts_code, buy_dt)
|
||||
ctx = {"chase": None, "day_chg": None, "mom5": None, "red_vol": None}
|
||||
if not bars:
|
||||
return ctx
|
||||
bd = buy_dt.date()
|
||||
bar = bars.get(bd)
|
||||
if bar is None:
|
||||
# 账本时间戳那天没有行情(极少见), 用其后第一根近似
|
||||
later = sorted(d for d in bars if d >= bd)
|
||||
if not later:
|
||||
return ctx
|
||||
bd = later[0]
|
||||
bar = bars[bd]
|
||||
hi, lo = bar["high"], bar["low"]
|
||||
if buy_price > 0 and hi > lo:
|
||||
ctx["chase"] = max(0.0, min(1.0, (buy_price - lo) / (hi - lo)))
|
||||
if bar["pct"]:
|
||||
ctx["day_chg"] = bar["pct"] / 100.0
|
||||
elif bar["pre_close"] > 0:
|
||||
ctx["day_chg"] = bar["close"] / bar["pre_close"] - 1.0
|
||||
prior = sorted((d, b) for d, b in bars.items() if d < bd)
|
||||
if len(prior) >= 5:
|
||||
c5 = prior[-5][1]["close"]
|
||||
if c5 > 0:
|
||||
ctx["mom5"] = bar["close"] / c5 - 1.0
|
||||
vols = [b["vol"] for _, b in prior[-5:] if b["vol"] > 0]
|
||||
if vols and bar["vol"] > 0:
|
||||
avgv = sum(vols) / len(vols)
|
||||
ctx["red_vol"] = (bar["close"] < bar["open"]) and (bar["vol"] >= REDVOL_VOL_MULT * avgv)
|
||||
return ctx
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ 成本
|
||||
def trade_cost(notional: float, is_sell: bool) -> float:
|
||||
n = abs(_f(notional))
|
||||
if n <= 0:
|
||||
return 0.0
|
||||
commission = max(n * COMMISSION_RATE, COMMISSION_MIN)
|
||||
stamp = n * STAMP_RATE if is_sell else 0.0
|
||||
transfer = n * TRANSFER_RATE
|
||||
slip = n * SLIPPAGE_BPS / 10000.0
|
||||
return commission + stamp + transfer + slip
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ 来回配对 (买→其后第一次卖)
|
||||
def pair_round_trips(rows_by_code: dict) -> list:
|
||||
"""同一只票: 把"一次买"和其后"第一次卖"配成一个来回(粗配, 不做逐笔 FIFO)。
|
||||
说明: 多次买(加仓)只认第一笔为入场; 入场时点体检看的是首次进场那一下。"""
|
||||
trips = []
|
||||
for code, rows in rows_by_code.items():
|
||||
pending_buy = None
|
||||
for r in rows:
|
||||
tag = classify(r)
|
||||
if tag["side"] == "buy":
|
||||
if pending_buy is None:
|
||||
pending_buy = r
|
||||
elif tag["side"] == "sell" and pending_buy is not None:
|
||||
b, s = pending_buy, r
|
||||
bt, st = b["decided_at"], s["decided_at"]
|
||||
bt = bt if isinstance(bt, datetime) else datetime.fromisoformat(str(bt))
|
||||
st = st if isinstance(st, datetime) else datetime.fromisoformat(str(st))
|
||||
hold_days = (st.date() - bt.date()).days
|
||||
gross = _pct(b["price_at"], s["price_at"])
|
||||
trips.append({
|
||||
"ts_code": code, "buy_at": bt, "sell_at": st,
|
||||
"buy_price": _f(b["price_at"]), "sell_price": _f(s["price_at"]),
|
||||
"hold_days": hold_days, "gross_ret": gross,
|
||||
"signal_driven": tag["signal_driven"], "cohort": tag.get("cohort"),
|
||||
"sell_reason": s.get("reason"),
|
||||
})
|
||||
pending_buy = None
|
||||
return trips
|
||||
|
||||
|
||||
def is_bad_entry(t: dict) -> bool:
|
||||
"""坏入场: 决策系统驱动的卖出、且持有不超过 QUICK_DAYS —— 买完就被风控快速清掉的那批。"""
|
||||
return bool(t["signal_driven"]) and t["hold_days"] is not None and t["hold_days"] <= QUICK_DAYS
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ 各段输出
|
||||
def section_inventory(rows, trips):
|
||||
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")
|
||||
bad = [t for t in trips if is_bad_entry(t)]
|
||||
print("\n① 样本盘点")
|
||||
print(f" 账本落地买入 {buys} 笔 · 落地卖出 {sells} 笔 · 配成来回 {len(trips)} 组")
|
||||
print(f" 其中「买完就被风控快速清掉」的坏入场(卖出为决策系统驱动、持有≤{QUICK_DAYS}交易日) "
|
||||
f"{len(bad)} 组")
|
||||
return bad
|
||||
|
||||
|
||||
def _profile(trips):
|
||||
"""算一批来回的入场画像三特征均值。返回 (chase, day_chg, mom5, n_ctx)。"""
|
||||
ch, dc, mo = [], [], []
|
||||
for t in trips:
|
||||
c = entry_context(t["ts_code"], t["buy_at"], t["buy_price"])
|
||||
ch.append(c["chase"]); dc.append(c["day_chg"]); mo.append(c["mom5"])
|
||||
n_ctx = sum(1 for x in ch if x is not None)
|
||||
return _avg(ch), _avg(dc), _avg(mo), n_ctx
|
||||
|
||||
|
||||
def section_profile(trips):
|
||||
"""② 坏入场画像 vs 全体入场: 三个买入日价格特征。"""
|
||||
print("\n② 坏入场画像 vs 全体入场 (买入日的价格情形)")
|
||||
all_paired = [t for t in trips if t["buy_price"] > 0]
|
||||
bad = [t for t in all_paired if is_bad_entry(t)]
|
||||
ac, ad, am, an = _profile(all_paired)
|
||||
bc, bd, bm, bn = _profile(bad)
|
||||
if an == 0:
|
||||
print(" (前向/日线价源未接通 fetch_daily_bars, 本段跳过 —— 接上后重跑即出)")
|
||||
return
|
||||
print(f" 全体入场 {len(all_paired)} 组(有行情 {an} 组): "
|
||||
f"追高度均值 {ac:.2f} · 买入日涨跌 {_fmt_pct(ad)} · 前5日动量 {_fmt_pct(am)}")
|
||||
if bn < MIN_SAMPLE:
|
||||
print(f" 坏入场 {len(bad)} 组(有行情 {bn} 组, <{MIN_SAMPLE}): 只列数不下结论。")
|
||||
print(f" 坏入场 {len(bad)} 组(有行情 {bn} 组): "
|
||||
f"追高度均值 {bc if bc is None else round(bc,2)} · 买入日涨跌 {_fmt_pct(bd)} · "
|
||||
f"前5日动量 {_fmt_pct(bm)}")
|
||||
print(" 读法: 追高度=买价在当日最高最低之间的位置(1=买在最高)。坏入场若追高度明显更高、"
|
||||
"或前5日动量明显更负, 说明它们多买在冲高或买在已下跌的票上 —— 那正是可在入场端拦的把手。")
|
||||
|
||||
|
||||
def _fwd_profile(trips):
|
||||
got = {h: [] for h in FWD_HORIZONS}
|
||||
for t in trips:
|
||||
fr = forward_returns(t["ts_code"], t["buy_at"], t["buy_price"])
|
||||
for h in FWD_HORIZONS:
|
||||
if fr[h] is not None:
|
||||
got[h].append(fr[h])
|
||||
return got
|
||||
|
||||
|
||||
def _print_fwd(label, got):
|
||||
if not any(got.values()):
|
||||
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}), 只报数")
|
||||
continue
|
||||
avg = sum(v) / len(v)
|
||||
down = sum(1 for x in v if x < 0) / len(v)
|
||||
print(f" T+{h}: 样本 {len(v)} · 买后平均 {_fmt_pct(avg)} · 买后就跌比例 {down:.0%}")
|
||||
return True
|
||||
|
||||
|
||||
def section_forward(trips):
|
||||
"""③ 入场后前向收益: 买入时点本身好不好 (坏入场 vs 全体)。"""
|
||||
print("\n③ 买入之后的前向收益 (相对买价; 负=买在了下跌前, 入场时点差)")
|
||||
all_paired = [t for t in trips if t["buy_price"] > 0]
|
||||
bad = [t for t in all_paired if is_bad_entry(t)]
|
||||
if not _print_fwd("全体入场", _fwd_profile(all_paired)):
|
||||
print(" (前向价源未接通 fetch_daily_bars, 本段跳过 —— 接上后重跑即出)")
|
||||
return
|
||||
_print_fwd("坏入场 (买完就被风控清掉这批)", _fwd_profile(bad))
|
||||
print(" 读法: 坏入场买后前向明显比全体更负=这批入场时点确实差(不是卖错是买错); "
|
||||
"若全体入场买后也普遍为负, 说明入场时点是系统性问题, 不止这十几笔。")
|
||||
|
||||
|
||||
def section_cost(trips):
|
||||
print("\n④ 坏入场的来回, 纯摩擦成本")
|
||||
bad = [t for t in trips if is_bad_entry(t) and t["gross_ret"] is not None]
|
||||
if not bad:
|
||||
print(" 无样本。")
|
||||
return
|
||||
total_rate = len(bad) * _rt_cost_rate()
|
||||
print(f" 每个来回往返摩擦约 {_rt_cost_rate():.3%} · {len(bad)} 组坏入场累计摩擦 ≈ "
|
||||
f"名义规模的 {total_rate:.2%}")
|
||||
print(" 读法: 这是「买错就得原路平掉」白交的过路费, 少买一笔坏入场就省一份。")
|
||||
|
||||
|
||||
# ---- ⑤ 反事实入场过滤 ----
|
||||
def _filters():
|
||||
"""候选入场闸门: 每条是 (名称, 说明, 判定函数(trip,ctx)->bool 命中即"该拦")。
|
||||
只用买入日及之前的信息, 无未来函数。"""
|
||||
return [
|
||||
("追高", f"买价落在当日振幅顶部{(1-CHASE_TOP_FRAC):.0%}以内(追高度≥{CHASE_TOP_FRAC})",
|
||||
lambda t, c: c["chase"] is not None and c["chase"] >= CHASE_TOP_FRAC),
|
||||
("逆势买", f"买入日已较前5日跌超{abs(DOWNTREND_5D):.0%}(前5日动量≤{DOWNTREND_5D:.0%})",
|
||||
lambda t, c: c["mom5"] is not None and c["mom5"] <= DOWNTREND_5D),
|
||||
("放量阴线", f"买入日是阴线且量能≥前5日均量×{REDVOL_VOL_MULT}",
|
||||
lambda t, c: c["red_vol"] is True),
|
||||
]
|
||||
|
||||
|
||||
def section_counterfactual(trips):
|
||||
"""⑤ 反事实: 在入场端加一道过滤, 净赚还是误伤好票。"""
|
||||
print("\n⑤ 反事实: 入场端加一道过滤, 净赚还是误伤好票")
|
||||
uni = [t for t in trips if t["buy_price"] > 0 and t["gross_ret"] is not None]
|
||||
# 预取每笔的入场情形, 顺带探价源
|
||||
ctxs = {id(t): entry_context(t["ts_code"], t["buy_at"], t["buy_price"]) for t in uni}
|
||||
if not any(c["chase"] is not None or c["mom5"] is not None for c in ctxs.values()):
|
||||
print(" (前向/日线价源未接通, 本段跳过 —— 接上 fetch_daily_bars 后重跑即出)")
|
||||
return
|
||||
n_bad = sum(1 for t in uni if is_bad_entry(t))
|
||||
print(f" 口径: 全体可判来回 {len(uni)} 组(其中坏入场 {n_bad} 组)。"
|
||||
f"每条过滤=不买命中的那些票, 省掉其整个来回的实现净收益(毛收益减往返摩擦{_rt_cost_rate():.2%})。")
|
||||
for name, desc, fn in _filters():
|
||||
blocked = [t for t in uni if fn(t, ctxs[id(t)])]
|
||||
if not blocked:
|
||||
print(f" · {name}({desc}): 一笔没命中, 跳过。")
|
||||
continue
|
||||
nets = [t["gross_ret"] - _rt_cost_rate() for t in blocked]
|
||||
caught_bad = sum(1 for t in blocked if is_bad_entry(t))
|
||||
winners = sum(1 for x in nets if x > 0)
|
||||
avg_net = sum(nets) / len(nets)
|
||||
portfolio = -sum(nets) # 不买这些 → 组合少了它们的净收益; 正=少亏(有用)
|
||||
recall = (caught_bad / n_bad) if n_bad else None
|
||||
if len(blocked) < MIN_SAMPLE:
|
||||
print(f" · {name}({desc}): 挡掉 {len(blocked)} 组(<{MIN_SAMPLE}, 只报数) · "
|
||||
f"含坏入场 {caught_bad} 组 · 挡掉里赚钱 {winners} 组")
|
||||
continue
|
||||
print(f" · {name}: 挡掉 {len(blocked)} 组 · 含坏入场 {caught_bad} 组"
|
||||
f"{'' if recall is None else f'(召回坏入场 {recall:.0%})'} · "
|
||||
f"挡掉平均实现净收益 {_fmt_pct(avg_net)} · 误伤率 {winners/len(blocked):.0%} · "
|
||||
f"组合净效果 {_fmt_pct(portfolio)}")
|
||||
print(f" ({desc})")
|
||||
print(" 读法: 某条过滤「挡掉平均实现净收益」为负(挡掉的多是亏钱来回)、误伤率低、组合净效果为正,"
|
||||
" 才值得上; 若误伤率高(挡掉里不少是赚的)或净效果为负, 说明它连好票一起误杀, 别上。"
|
||||
" 召回=这条能拦住多少比例的坏入场。")
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ main
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description="入场体检: 坏入场画像与入场过滤反事实 (只读)")
|
||||
ap.add_argument("--days", type=int, default=120, help="回看多少自然日 (默认120)")
|
||||
ap.add_argument("--since", type=str, default=None, help="或指定起始日 YYYY-MM-DD")
|
||||
args = ap.parse_args()
|
||||
|
||||
since = (datetime.strptime(args.since, "%Y-%m-%d") if args.since
|
||||
else datetime.now() - timedelta(days=args.days))
|
||||
global _ANALYSIS_START, _ANALYSIS_END
|
||||
_ANALYSIS_START, _ANALYSIS_END = since, datetime.now()
|
||||
print(f"入场体检 · 账本自 {since:%Y-%m-%d} 起 · 成本口径 往返≈{_rt_cost_rate():.3%} · "
|
||||
f"坏入场=决策系统卖出且持有≤{QUICK_DAYS}交易日")
|
||||
|
||||
try:
|
||||
rows = read_ledger(since)
|
||||
except DBUnavailable as e:
|
||||
print(f"[FAIL] 读账本失败(库不可达): {e}"); sys.exit(1)
|
||||
if not rows:
|
||||
print("账本在该窗口内为空 —— 换个更长的 --days 再看。"); return
|
||||
|
||||
by_code = {}
|
||||
for r in rows:
|
||||
by_code.setdefault(r["ts_code"], []).append(r)
|
||||
trips = pair_round_trips(by_code)
|
||||
|
||||
section_inventory(rows, trips)
|
||||
section_profile(trips)
|
||||
section_forward(trips)
|
||||
section_cost(trips)
|
||||
section_counterfactual(trips)
|
||||
print("\n完成。②③⑤ 若显示「未接通」, 是历史日线源(fetch_daily_bars)还没接 —— "
|
||||
"确认接哪张表后重跑即全。")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
Reference in New Issue