202 lines
9.1 KiB
Python
202 lines
9.1 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
第四批模块单测 (实机运行, 零外部依赖)
|
||
======================================
|
||
运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch4_units.py
|
||
覆盖: action_engine 四类自主动作的触发边界与数量口径 (FILL/ADD/DCA/TRIM) 及扫描剪枝。
|
||
"""
|
||
import os
|
||
import sys
|
||
import traceback
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from app.core import action_engine as ae # noqa: E402
|
||
|
||
RESULTS = []
|
||
|
||
|
||
def case(name):
|
||
def deco(fn):
|
||
RESULTS.append((name, fn))
|
||
return fn
|
||
return deco
|
||
|
||
|
||
PARAMS = {"scale": 2_000_000, "stock_target_default": 0.06,
|
||
"batch_split": (0.5, 0.25, 0.25), "cushion_solid": 0.03,
|
||
"trim_peak": 0.06, "trim_giveback": 0.5,
|
||
"dca_triggers": (-0.08, -0.15), "dca_deep_confirm": -0.15, "dca_max_ratio": 0.5,
|
||
"no_chase_ma5": 0.06, "build_window_tdays": 10, "fill_max_loss": -0.03}
|
||
|
||
|
||
def pos(**kw):
|
||
p = {"ts_code": "600000.SH", "price": 10.0, "avg_cost": 10.0, "total_qty": 6000,
|
||
"base_qty": 6000, "add_qty": 0, "dca_qty": 0, "market_value": 60_000,
|
||
"cushion_pct": 0.0, "cushion_peak": 0.0, "target_pct": 0.06,
|
||
"support_ref": None, "pressure_ref": None, "stop_ref": None,
|
||
"fill_count": 0, "dca_count": 0, "frozen_reason": "NONE"}
|
||
p.update(kw)
|
||
return p
|
||
|
||
|
||
def mkt(**kw):
|
||
m = {"ma5": 10.0, "high5": 10.0, "tdays_since_open": 3, "tdays_since_last_add": 5}
|
||
m.update(kw)
|
||
return m
|
||
|
||
|
||
# ================================================================ FILL
|
||
@case("回踩补足·建仓期内浅亏未破支撑 → 按补足批额度补到目标")
|
||
def _():
|
||
c = ae.eval_fill(pos(price=9.8, cushion_pct=-0.02, support_ref=9.7, market_value=58_800),
|
||
PARAMS, mkt())
|
||
assert c and c["action"] == "FILL" and c["side"] == "buy", c
|
||
assert c["qty"] == 3000, c # 补足批 3 万 ÷ 9.8 → 3000 股
|
||
assert c["judge_required"] and not c["needs_user_confirm"]
|
||
assert "未破支撑" in c["reason"]
|
||
|
||
|
||
@case("回踩补足·每票一次 / 出建仓期 / 亏太深 / 破支撑 / 无支撑 一律不提")
|
||
def _():
|
||
base = dict(price=9.8, cushion_pct=-0.02, support_ref=9.7, market_value=58_800)
|
||
assert ae.eval_fill(pos(**base, fill_count=1), PARAMS, mkt()) is None
|
||
assert ae.eval_fill(pos(**base), PARAMS, mkt(tdays_since_open=11)) is None
|
||
assert ae.eval_fill(pos(price=9.5, cushion_pct=-0.05, support_ref=9.0,
|
||
market_value=57_000), PARAMS, mkt()) is None
|
||
assert ae.eval_fill(pos(price=9.5, cushion_pct=-0.02, support_ref=9.7,
|
||
market_value=57_000), PARAMS, mkt()) is None # 破支撑
|
||
assert ae.eval_fill(pos(**{**base, "support_ref": None}), PARAMS, mkt()) is None
|
||
# 已转盈就不是「回踩补足」的场景了
|
||
assert ae.eval_fill(pos(price=10.5, cushion_pct=0.05, support_ref=9.7,
|
||
market_value=63_000), PARAMS, mkt()) is None
|
||
|
||
|
||
# ================================================================ ADD
|
||
@case("盈利加仓·厚垫 + 创5日新高 → 按加仓批额度加")
|
||
def _():
|
||
c = ae.eval_add(pos(price=11.0, cushion_pct=0.10, market_value=66_000),
|
||
PARAMS, mkt(ma5=10.6, high5=11.0))
|
||
assert c and c["action"] == "ADD" and c["qty"] == 2700, c # 3万 ÷ 11 → 2700 股
|
||
assert "创 5 日新高" in c["reason"], c
|
||
# 站上压力位也算
|
||
c2 = ae.eval_add(pos(price=11.0, cushion_pct=0.10, pressure_ref=10.9,
|
||
market_value=66_000), PARAMS, mkt(ma5=10.6, high5=12.0))
|
||
assert c2 and "压力位" in c2["reason"], c2
|
||
|
||
|
||
@case("盈利加仓·薄垫/未突破/距上次<2日/追高 一律不提")
|
||
def _():
|
||
assert ae.eval_add(pos(price=11.0, cushion_pct=0.02, market_value=66_000),
|
||
PARAMS, mkt(high5=11.0)) is None # 垫子不厚
|
||
assert ae.eval_add(pos(price=10.5, cushion_pct=0.10, market_value=63_000),
|
||
PARAMS, mkt(high5=12.0)) is None # 没新高没压力位
|
||
assert ae.eval_add(pos(price=11.0, cushion_pct=0.10, market_value=66_000),
|
||
PARAMS, mkt(high5=11.0, tdays_since_last_add=1)) is None
|
||
assert ae.eval_add(pos(price=11.0, cushion_pct=0.10, market_value=66_000),
|
||
PARAMS, mkt(ma5=10.0, high5=11.0)) is None # 距 MA5 10% > 6%
|
||
|
||
|
||
# ================================================================ DCA
|
||
@case("补仓·触及 -8% 首档 → 按底仓一半提, 不强制确认")
|
||
def _():
|
||
c = ae.eval_dca(pos(price=9.1, cushion_pct=-0.09, market_value=54_600), PARAMS, mkt())
|
||
assert c and c["action"] == "DCA" and c["qty"] == 3000, c # 底仓 6000 × 50%
|
||
assert c["needs_user_confirm"] is False and c["judge_required"], c
|
||
assert c["hard_numbers"]["stage"] == 1
|
||
|
||
|
||
@case("补仓·-15% 及更深 永远需用户确认 + 研判须答杀逻辑还是杀情绪")
|
||
def _():
|
||
c = ae.eval_dca(pos(price=8.4, cushion_pct=-0.16, market_value=50_400), PARAMS, mkt())
|
||
assert c and c["needs_user_confirm"] is True, c
|
||
assert c["hard_numbers"]["stage"] == 2 and "必须用户确认" in c["reason"], c
|
||
|
||
|
||
@case("补仓·各档只评估一次 / 终身只执行一次 / 浮盈不评估")
|
||
def _():
|
||
assert ae.eval_dca(pos(cushion_pct=-0.09, dca_count=1), PARAMS, mkt()) is None
|
||
assert ae.eval_dca(pos(cushion_pct=-0.16, dca_count=1), PARAMS, mkt()) is not None
|
||
assert ae.eval_dca(pos(cushion_pct=-0.16, dca_count=2), PARAMS, mkt()) is None
|
||
assert ae.eval_dca(pos(cushion_pct=-0.20, dca_qty=1000), PARAMS, mkt()) is None
|
||
assert ae.eval_dca(pos(cushion_pct=-0.05), PARAMS, mkt()) is None
|
||
assert ae.eval_dca(pos(cushion_pct=0.05), PARAMS, mkt()) is None
|
||
|
||
|
||
# ================================================================ TRIM
|
||
@case("保垫减仓·峰值≥6%且回吐过半 → 减 1/3, 不需研判不需确认")
|
||
def _():
|
||
c = ae.eval_trim(pos(cushion_peak=0.08, cushion_pct=0.04), PARAMS)
|
||
assert c and c["action"] == "TRIM" and c["side"] == "sell", c
|
||
assert c["qty"] == 2000, c # 6000 的 1/3
|
||
assert not c["judge_required"] and not c["needs_user_confirm"], c
|
||
|
||
|
||
@case("保垫减仓·峰值不足或回吐不到一半 不提")
|
||
def _():
|
||
assert ae.eval_trim(pos(cushion_peak=0.05, cushion_pct=-0.01), PARAMS) is None
|
||
assert ae.eval_trim(pos(cushion_peak=0.08, cushion_pct=0.0401), PARAMS) is None
|
||
assert ae.eval_trim(pos(cushion_peak=0.08, cushion_pct=None), PARAMS) is None
|
||
assert ae.eval_trim(pos(cushion_peak=0.08, cushion_pct=0.04, total_qty=200),
|
||
PARAMS) is None # 1/3 不足一手
|
||
|
||
|
||
# ================================================================ 扫描
|
||
@case("扫描·冻结票只评减仓 / 已在途不重复提 / 单票异常不拖垮整轮")
|
||
def _():
|
||
ps = [pos(ts_code="600000.SH", cushion_peak=0.08, cushion_pct=0.04,
|
||
frozen_reason="COMMAND_HALT"),
|
||
pos(ts_code="000001.SZ", price=11.0, cushion_pct=0.10, market_value=66_000),
|
||
pos(ts_code="600519.SH", price=None, cushion_pct=0.10, market_value=66_000)]
|
||
r = ae.scan(positions=ps, params=PARAMS, market={
|
||
"600000.SH": mkt(), "000001.SZ": mkt(ma5=10.6, high5=11.0), "600519.SH": mkt()})
|
||
acts = {(c["ts_code"], c["action"]) for c in r["candidates"]}
|
||
assert ("600000.SH", "TRIM") in acts, acts # 冻结不挡减仓
|
||
assert not any(c[0] == "600000.SH" and c[1] != "TRIM" for c in acts), acts
|
||
assert ("000001.SZ", "ADD") in acts, acts
|
||
assert any(s["why"].endswith("禁增持") for s in r["skipped"]), r["skipped"]
|
||
|
||
r2 = ae.scan(positions=ps, params=PARAMS,
|
||
market={"600000.SH": mkt(), "000001.SZ": mkt(ma5=10.6, high5=11.0),
|
||
"600519.SH": mkt()},
|
||
skip={("000001.SZ", "ADD")})
|
||
assert ("000001.SZ", "ADD") not in {(c["ts_code"], c["action"]) for c in r2["candidates"]}
|
||
assert any(s.get("why") == "已有在途提议/指令" for s in r2["skipped"]), r2["skipped"]
|
||
|
||
# 空仓票直接跳过, 不进候选
|
||
r3 = ae.scan(positions=[pos(ts_code="300750.SZ", total_qty=0)], params=PARAMS, market={})
|
||
assert r3["candidates"] == []
|
||
|
||
|
||
@case("扫描·批次额度与距目标空间口径")
|
||
def _():
|
||
p = pos(market_value=100_000)
|
||
assert ae.batch_amount(p, PARAMS, 0) == 60_000.0 # 底仓 50%
|
||
assert ae.batch_amount(p, PARAMS, 1) == 30_000.0 # 补足 25%
|
||
assert ae.batch_amount(p, PARAMS, 2) == 30_000.0 # 加仓 25%
|
||
assert ae.room_to_target(p, PARAMS) == 20_000.0
|
||
assert ae.room_to_target(pos(market_value=130_000), PARAMS) == 0.0
|
||
|
||
|
||
# ---------------------------------------------------------------- runner
|
||
def main():
|
||
passed, failed = 0, 0
|
||
for name, fn in RESULTS:
|
||
try:
|
||
fn()
|
||
print(f" PASS {name}")
|
||
passed += 1
|
||
except Exception:
|
||
print(f" FAIL {name}")
|
||
traceback.print_exc()
|
||
failed += 1
|
||
print("-" * 60)
|
||
if failed:
|
||
print(f"FAILED: {failed} / {passed + failed}")
|
||
sys.exit(1)
|
||
print(f"ALL PASS ({passed} cases)")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|