219 lines
8.8 KiB
Python
219 lines
8.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""三源合议 · 工作包三离场纪律 (2026-09-11 方案第五节, 台账 010)。全部离线, 不连库。
|
|
|
|
入场靠投票, 离场靠纪律: SAR 转空像保垫减仓一样按规则执行, 不进入场那套合议投票。
|
|
A eval_tech_exit: 相位转空才评/确认清仓·未确认减三分之一/翻向超期不评/同一翻空一次/
|
|
无读数弃权/propose_only 交人/可卖量夹紧/不足一手不评/开关关掉逐字回旧。
|
|
B 卖出优先级: 目标价 < 研究走弱 < 技术面转空 < 保垫减仓。
|
|
C scan 端到端: 产出来源 tech_exit / 同轮与保垫减仓并现留转空 / 目标价优先于转空。
|
|
D 策略票只看目标价 (按函数判, 不按动作名) —— 技术面转空对策略票不评。
|
|
E 参数登记。
|
|
"""
|
|
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
|
|
from app.services import param_store as ps # noqa: E402
|
|
|
|
RESULTS = []
|
|
|
|
|
|
def case(name):
|
|
def deco(fn):
|
|
RESULTS.append((name, fn))
|
|
return fn
|
|
return deco
|
|
|
|
|
|
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", "avail_qty": 6000}
|
|
p.update(kw)
|
|
return p
|
|
|
|
|
|
def tech(phase="转空", flip=1, confirm=False, sar_value=9.5, reason="SAR 翻空"):
|
|
return {"phase": phase, "sar_flip_days": flip, "confirm": confirm,
|
|
"sar_value": sar_value, "reason": reason, "stance": "看空"}
|
|
|
|
|
|
def tpos(t, **kw):
|
|
p = pos(**kw)
|
|
p["tech"] = t
|
|
return p
|
|
|
|
|
|
def tparams(**kw):
|
|
p = {"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,
|
|
"tech_exit_on": True, "tech_exit_propose_only": False,
|
|
"tech_exit_trim_ratio": 1.0 / 3, "tech_exit_fresh_days": 2, "tech_exit_done": set()}
|
|
p.update(kw)
|
|
return p
|
|
|
|
|
|
# ================================================================ A eval_tech_exit
|
|
@case("A 转空离场·开关关掉逐字回旧 (tech_exit_on 假 → 不评)")
|
|
def _():
|
|
assert ae.eval_tech_exit(tpos(tech(confirm=True)), tparams(tech_exit_on=False)) is None
|
|
|
|
|
|
@case("A 转空离场·相位非转空不评")
|
|
def _():
|
|
assert ae.eval_tech_exit(tpos(tech(phase="趋势空", confirm=True)), tparams()) is None
|
|
assert ae.eval_tech_exit(tpos(tech(phase="趋势多", confirm=False)), tparams()) is None
|
|
|
|
|
|
@case("A 转空离场·确认转空清仓全部可卖量 (EXIT, 来源 tech_exit)")
|
|
def _():
|
|
c = ae.eval_tech_exit(tpos(tech(confirm=True), total_qty=6000, avail_qty=6000), tparams())
|
|
assert c and c["action"] == "EXIT" and c["side"] == "sell"
|
|
assert c["qty"] == 6000 and c["source"] == ae.SRC_TECH_EXIT
|
|
|
|
|
|
@case("A 转空离场·未确认减三分之一 (TRIM)")
|
|
def _():
|
|
c = ae.eval_tech_exit(tpos(tech(confirm=False), total_qty=6000, avail_qty=6000), tparams())
|
|
assert c and c["action"] == "TRIM" and c["qty"] == 2000
|
|
|
|
|
|
@case("A 转空离场·翻向超期不评 (超过 fresh_days)")
|
|
def _():
|
|
assert ae.eval_tech_exit(tpos(tech(flip=3, confirm=True)), tparams()) is None
|
|
assert ae.eval_tech_exit(tpos(tech(flip=2, confirm=True)), tparams()) is not None
|
|
|
|
|
|
@case("A 转空离场·同一翻空已处理过不再评")
|
|
def _():
|
|
p = tpos(tech(confirm=True), ts_code="600000.SH")
|
|
assert ae.eval_tech_exit(p, tparams(tech_exit_done={"600000.SH"})) is None
|
|
assert ae.eval_tech_exit(p, tparams(tech_exit_done=set())) is not None
|
|
|
|
|
|
@case("A 转空离场·无技术面读数不评 (无读数弃权)")
|
|
def _():
|
|
p = pos()
|
|
p.pop("tech", None)
|
|
assert ae.eval_tech_exit(p, tparams()) is None
|
|
p2 = pos()
|
|
p2["tech"] = None
|
|
assert ae.eval_tech_exit(p2, tparams()) is None
|
|
|
|
|
|
@case("A 转空离场·propose_only 打交人标记")
|
|
def _():
|
|
c = ae.eval_tech_exit(tpos(tech(confirm=True), total_qty=6000, avail_qty=6000),
|
|
tparams(tech_exit_propose_only=True))
|
|
assert c and c.get("needs_user_confirm") is True and "propose_only" in (c.get("confirm_why") or "")
|
|
|
|
|
|
@case("A 转空离场·按 T+1 可卖量夹紧 (确认时清可卖量而非总持仓)")
|
|
def _():
|
|
c = ae.eval_tech_exit(tpos(tech(confirm=True), total_qty=6000, avail_qty=3000), tparams())
|
|
assert c and c["qty"] == 3000
|
|
|
|
|
|
@case("A 转空离场·不足一手不评")
|
|
def _():
|
|
assert ae.eval_tech_exit(tpos(tech(confirm=True), total_qty=50, avail_qty=50), tparams()) is None
|
|
|
|
|
|
# ================================================================ B 卖出优先级
|
|
@case("B 卖出优先级·目标价 < 研究走弱 < 技术面转空 < 保垫减仓")
|
|
def _():
|
|
pr = lambda a, s: ae._sell_priority({"action": a, "source": s})
|
|
assert (pr("EXIT", ae.SRC_TARGET_PRICE) < pr("TRIM", ae.SRC_RESEARCH_WEAK)
|
|
< pr("EXIT", ae.SRC_TECH_EXIT) < pr("TRIM", ae.SRC_ENGINE))
|
|
|
|
|
|
# ================================================================ C scan 端到端
|
|
@case("C scan·确认转空产出 EXIT 候选")
|
|
def _():
|
|
p = tpos(tech(confirm=True), total_qty=6000, avail_qty=6000)
|
|
r = ae.scan(positions=[p], params=tparams(), market={p["ts_code"]: {}})
|
|
exits = [c for c in r["candidates"] if c.get("source") == ae.SRC_TECH_EXIT]
|
|
assert len(exits) == 1 and exits[0]["action"] == "EXIT"
|
|
|
|
|
|
@case("C scan·同轮与保垫减仓并现留技术面转空 (优先级更高)")
|
|
def _():
|
|
p = tpos(tech(confirm=True), total_qty=6000, avail_qty=6000,
|
|
cushion_peak=0.08, cushion_pct=0.04)
|
|
r = ae.scan(positions=[p], params=tparams(), market={p["ts_code"]: {}})
|
|
sells = [c for c in r["candidates"] if c["side"] == "sell"]
|
|
assert len(sells) == 1 and sells[0]["source"] == ae.SRC_TECH_EXIT
|
|
|
|
|
|
@case("C scan·目标价到价优先于技术面转空")
|
|
def _():
|
|
p = tpos(tech(confirm=True), total_qty=6000, avail_qty=6000, price=13.0)
|
|
r = ae.scan(positions=[p], params=tparams(), market={p["ts_code"]: {}},
|
|
stock_params={"600000.SH": {"target_price": 12.5}})
|
|
sells = [c for c in r["candidates"] if c["side"] == "sell"]
|
|
assert len(sells) == 1 and sells[0]["source"] == ae.SRC_TARGET_PRICE
|
|
|
|
|
|
@case("C scan·开关关掉逐字回旧: 挂了转空读数也不产出转空离场")
|
|
def _():
|
|
p = tpos(tech(confirm=True), total_qty=6000, avail_qty=6000)
|
|
r = ae.scan(positions=[p], params=tparams(tech_exit_on=False), market={p["ts_code"]: {}})
|
|
assert not any(c.get("source") == ae.SRC_TECH_EXIT for c in r["candidates"])
|
|
|
|
|
|
# ================================================================ D 策略票只看目标价 (Part 2)
|
|
@case("D 策略票·目标价照评但技术面转空不评 (按函数判)")
|
|
def _():
|
|
p = tpos(tech(confirm=True), total_qty=6000, avail_qty=6000, price=13.0)
|
|
r = ae.scan(positions=[p], params=tparams(), market={p["ts_code"]: {}},
|
|
strategy_codes={"600000.SH"}, stock_params={"600000.SH": {"target_price": 12.5}})
|
|
assert any(c.get("source") == ae.SRC_TARGET_PRICE for c in r["candidates"]), "策略票目标价照评"
|
|
assert not any(c.get("source") == ae.SRC_TECH_EXIT for c in r["candidates"]), "策略票技术面转空不评"
|
|
|
|
|
|
@case("D 策略票·无目标价时技术面转空也不产出 (策略层接管)")
|
|
def _():
|
|
p = tpos(tech(confirm=True), total_qty=6000, avail_qty=6000)
|
|
r = ae.scan(positions=[p], params=tparams(), market={p["ts_code"]: {}},
|
|
strategy_codes={"600000.SH"})
|
|
assert not any(c.get("source") == ae.SRC_TECH_EXIT for c in r["candidates"])
|
|
|
|
|
|
# ================================================================ E 参数登记
|
|
@case("E 参数·转空离场三键登记, 档位默认 full")
|
|
def _():
|
|
for k in ("PMS_TECH_EXIT_AUTONOMY", "PMS_TECH_EXIT_TRIM_RATIO", "PMS_TECH_EXIT_DONE"):
|
|
assert k in ps.RUNTIME_EXTRA, k
|
|
assert ps.RUNTIME_EXTRA["PMS_TECH_EXIT_AUTONOMY"][0] == "full"
|
|
assert ps.RUNTIME_EXTRA["PMS_TECH_EXIT_AUTONOMY"][1] is str
|
|
assert abs(ps.RUNTIME_EXTRA["PMS_TECH_EXIT_TRIM_RATIO"][0] - 1.0 / 3) < 1e-3
|
|
|
|
|
|
def main():
|
|
ok = 0
|
|
for name, fn in RESULTS:
|
|
try:
|
|
fn()
|
|
ok += 1
|
|
print(" ok " + name)
|
|
except Exception:
|
|
print(" FAIL " + name)
|
|
traceback.print_exc()
|
|
print("-" * 60)
|
|
if ok == len(RESULTS):
|
|
print("ALL PASS (%d cases)" % ok)
|
|
return 0
|
|
print("FAILED %d/%d" % (len(RESULTS) - ok, len(RESULTS)))
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|