434 lines
18 KiB
Python
434 lines
18 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
|
||
|
||
|
||
# ================================================================ F 跟踪止盈盘中 SAR 止损线 (part 3)
|
||
def _trail_sar(price, sar_line, now, avail=6000, avg=10.0, sold_day=0):
|
||
from app.services import strategy_runner as srun
|
||
st = {"ts_code": "600000.SH",
|
||
"params": {"giveback": 0.05, "sell_ratio": 1.0, "sar_line": sar_line}}
|
||
pos = {"avg_cost": avg, "avail_qty": avail, "total_qty": avail, "cushion_pct": price / avg - 1}
|
||
ctx = {"state": {"armed": False, "high_water": price, "sar_stop_day": sold_day}, "notes": []}
|
||
d = srun._eval_trail(st, pos, {"price": price}, now, ctx)
|
||
return d, ctx
|
||
|
||
|
||
@case("F SAR 线·09:45 后跌破 SAR 缓冲即全清 (leg=trail_sar, 当日记一次)")
|
||
def _():
|
||
from datetime import datetime
|
||
from app.services import strategy_runner as srun
|
||
d, ctx = _trail_sar(9.9, 10.0, datetime(2026, 9, 11, 10, 0)) # 9.9 < 10×0.997=9.97
|
||
assert d and d["action"] == srun.A_EXIT and d["leg"] == "trail_sar" and d["qty"] == 6000
|
||
from app.core import tradedays as td
|
||
assert ctx["state"]["sar_stop_day"] == td.ymd(datetime(2026, 9, 11, 10, 0))
|
||
|
||
|
||
@case("F SAR 线·09:45 前不触发")
|
||
def _():
|
||
from datetime import datetime
|
||
d, _ = _trail_sar(9.5, 10.0, datetime(2026, 9, 11, 9, 40))
|
||
assert d is None
|
||
|
||
|
||
@case("F SAR 线·现价在缓冲之上不触发")
|
||
def _():
|
||
from datetime import datetime
|
||
d, _ = _trail_sar(9.98, 10.0, datetime(2026, 9, 11, 10, 0)) # 9.98 > 9.97
|
||
assert d is None
|
||
|
||
|
||
@case("F SAR 线·当日只触发一次 (已记 sar_stop_day)")
|
||
def _():
|
||
from datetime import datetime
|
||
from app.core import tradedays as td
|
||
now = datetime(2026, 9, 11, 10, 0)
|
||
d, _ = _trail_sar(9.5, 10.0, now, sold_day=td.ymd(now))
|
||
assert d is None
|
||
|
||
|
||
@case("F SAR 线·没刷进 sar_line 时不判 (加不改, 拿不到 SAR 绝不当跌破)")
|
||
def _():
|
||
from datetime import datetime
|
||
from app.services import strategy_runner as srun
|
||
st = {"ts_code": "600000.SH", "params": {"giveback": 0.05, "sell_ratio": 1.0}} # 无 sar_line
|
||
pos = {"avg_cost": 10.0, "avail_qty": 6000, "total_qty": 6000, "cushion_pct": -0.05}
|
||
ctx = {"state": {"armed": False, "high_water": 9.5}, "notes": []}
|
||
assert srun._eval_trail(st, pos, {"price": 9.5}, datetime(2026, 9, 11, 10, 0), ctx) is None
|
||
|
||
|
||
@case("F SAR 线·开关关掉不触发")
|
||
def _():
|
||
from datetime import datetime
|
||
saved = ps.get_bool
|
||
ps.get_bool = lambda k, d=False: (False if k == "PMS_TECH_SAR_STOP_ON_TRAIL" else saved(k, d))
|
||
try:
|
||
d, _ = _trail_sar(9.5, 10.0, datetime(2026, 9, 11, 10, 0))
|
||
assert d is None
|
||
finally:
|
||
ps.get_bool = saved
|
||
|
||
|
||
# ================================================================ G SAR 线刷新腿 (part 3)
|
||
@case("G 刷新腿·只动自动挂载的 ACTIVE 跟踪止盈, 网格与人工不碰")
|
||
def _():
|
||
from app.services import strategy_advisor as adv, tech_service
|
||
from app.repo import pms_repo
|
||
calls = []
|
||
s_sm, s_up = tech_service.state_map, pms_repo.update_strategy
|
||
tech_service.state_map = lambda: {"600000.SH": {"sar_value": 9.55}}
|
||
pms_repo.update_strategy = lambda sid, **kw: (calls.append((sid, kw)) or 1)
|
||
try:
|
||
strat_all = [
|
||
{"strategy_id": "A", "ts_code": "600000.SH", "type": "TRAIL", "status": "ACTIVE",
|
||
"note": "自动挂载: x", "params": {"giveback": 0.05}},
|
||
{"strategy_id": "B", "ts_code": "600000.SH", "type": "GRID", "status": "ACTIVE",
|
||
"note": "自动挂载: x", "params": {}},
|
||
{"strategy_id": "C", "ts_code": "600000.SH", "type": "TRAIL", "status": "ACTIVE",
|
||
"note": "手工挂的", "params": {}},
|
||
]
|
||
out = {"errors": []}
|
||
adv._refresh_sar_lines(strat_all, out)
|
||
assert len(calls) == 1 and calls[0][0] == "A"
|
||
assert abs(calls[0][1]["params"]["sar_line"] - 9.55) < 1e-6
|
||
finally:
|
||
tech_service.state_map, pms_repo.update_strategy = s_sm, s_up
|
||
|
||
|
||
@case("G 刷新腿·无读数撤掉旧 SAR 线 (不拿旧读数当今天)")
|
||
def _():
|
||
from app.services import strategy_advisor as adv, tech_service
|
||
from app.repo import pms_repo
|
||
calls = []
|
||
s_sm, s_up = tech_service.state_map, pms_repo.update_strategy
|
||
tech_service.state_map = lambda: {} # 全无读数
|
||
pms_repo.update_strategy = lambda sid, **kw: (calls.append((sid, kw)) or 1)
|
||
try:
|
||
strat_all = [{"strategy_id": "A", "ts_code": "600000.SH", "type": "TRAIL",
|
||
"status": "ACTIVE", "note": "自动挂载: x",
|
||
"params": {"giveback": 0.05, "sar_line": 9.9}}]
|
||
out = {"errors": []}
|
||
adv._refresh_sar_lines(strat_all, out)
|
||
assert len(calls) == 1 and "sar_line" not in calls[0][1]["params"]
|
||
finally:
|
||
tech_service.state_map, pms_repo.update_strategy = s_sm, s_up
|
||
|
||
|
||
@case("G 参数·SAR 止损线两键登记")
|
||
def _():
|
||
assert ps.RUNTIME_EXTRA["PMS_TECH_SAR_STOP_ON_TRAIL"][0] is True
|
||
assert abs(ps.RUNTIME_EXTRA["PMS_TECH_SAR_STOP_BUFFER"][0] - 0.003) < 1e-9
|
||
|
||
|
||
# ================================================================ H 弱基本面试探仓紧止盈自动挂载 (part 4)
|
||
@case("H 紧止盈·入场账本 advice.tight_trail 为真才认")
|
||
def _():
|
||
from app.services import strategy_advisor as adv
|
||
from app.repo import pms_repo
|
||
s_lots, s_ins, s_led = pms_repo.list_lots, pms_repo.get_instruction, pms_repo.ledger_by_ref
|
||
pms_repo.list_lots = lambda code, **kw: [{"instruction_id": "I1"}]
|
||
pms_repo.get_instruction = lambda iid: {"origin_id": "P1"}
|
||
try:
|
||
pms_repo.ledger_by_ref = lambda refs: [{"verdict": "PASS",
|
||
"hard_numbers": {"advice": {"tight_trail": True}}}]
|
||
assert adv._entry_tight_trail("600000.SH") is True
|
||
pms_repo.ledger_by_ref = lambda refs: [{"verdict": "PASS",
|
||
"hard_numbers": {"advice": {"tight_trail": False}}}]
|
||
assert adv._entry_tight_trail("600000.SH") is False
|
||
pms_repo.ledger_by_ref = lambda refs: [{"verdict": "PASS", "hard_numbers": {}}]
|
||
assert adv._entry_tight_trail("600000.SH") is False
|
||
finally:
|
||
pms_repo.list_lots, pms_repo.get_instruction, pms_repo.ledger_by_ref = s_lots, s_ins, s_led
|
||
|
||
|
||
@case("H 紧止盈·非试探仓不挂 (返回 False, 走常规边)")
|
||
def _():
|
||
from app.services import strategy_advisor as adv
|
||
s = adv._entry_tight_trail
|
||
adv._entry_tight_trail = lambda code: False
|
||
try:
|
||
out = {"attached": [], "skipped": [], "blocked": []}
|
||
r = adv._attach_tight_trail({"ts_code": "600000.SH"}, {"optout_cooldown_tdays": 10},
|
||
set(), set(), True, out, None)
|
||
assert r is False and not out["attached"]
|
||
finally:
|
||
adv._entry_tight_trail = s
|
||
|
||
|
||
@case("H 紧止盈·试探仓 dry_run 挂 回撤3%/硬目标8%/带SAR线")
|
||
def _():
|
||
from app.services import strategy_advisor as adv, tech_service
|
||
s_e, s_sm = adv._entry_tight_trail, tech_service.state_map
|
||
adv._entry_tight_trail = lambda code: True
|
||
tech_service.state_map = lambda: {"600000.SH": {"sar_value": 9.6}}
|
||
try:
|
||
out = {"attached": [], "skipped": [], "blocked": []}
|
||
r = adv._attach_tight_trail({"ts_code": "600000.SH", "price": 10.0},
|
||
{"optout_cooldown_tdays": 10}, set(), set(), True, out, None)
|
||
assert r is True and len(out["attached"]) == 1
|
||
pr = out["attached"][0]["params"]
|
||
assert abs(pr["giveback"] - 0.03) < 1e-9 and abs(pr["hard_target"] - 0.08) < 1e-9
|
||
assert abs(pr["sar_line"] - 9.6) < 1e-6
|
||
finally:
|
||
adv._entry_tight_trail, tech_service.state_map = s_e, s_sm
|
||
|
||
|
||
@case("H 紧止盈·真挂调 strategy_service.attach 且 note 标弱基本面")
|
||
def _():
|
||
from app.services import strategy_advisor as adv, tech_service
|
||
from app.repo import pms_repo
|
||
s_e, s_sm, s_il = adv._entry_tight_trail, tech_service.state_map, pms_repo.insert_ledger
|
||
adv._entry_tight_trail = lambda code: True
|
||
tech_service.state_map = lambda: {}
|
||
pms_repo.insert_ledger = lambda **kw: "L"
|
||
calls = []
|
||
|
||
class SS:
|
||
def attach(self, spec, by=None):
|
||
calls.append(spec)
|
||
return {"ok": True, "strategy_id": "T9"}
|
||
try:
|
||
out = {"attached": [], "skipped": [], "blocked": [], "errors": []}
|
||
r = adv._attach_tight_trail({"ts_code": "600000.SH", "price": 10.0},
|
||
{"optout_cooldown_tdays": 10}, set(), set(), False, out, SS())
|
||
assert r is True and calls and calls[0]["type"] == "TRAIL"
|
||
assert "试探仓" in calls[0]["note"] and "紧止盈" in calls[0]["note"]
|
||
assert out["attached"][0]["strategy_id"] == "T9"
|
||
finally:
|
||
adv._entry_tight_trail, tech_service.state_map, pms_repo.insert_ledger = s_e, s_sm, s_il
|
||
|
||
|
||
@case("H 紧止盈·人工撤下冷却期内占位让路 (返回 True 不走常规边, 不挂)")
|
||
def _():
|
||
from app.services import strategy_advisor as adv
|
||
s_e = adv._entry_tight_trail
|
||
adv._entry_tight_trail = lambda code: True
|
||
try:
|
||
out = {"attached": [], "skipped": [], "blocked": []}
|
||
r = adv._attach_tight_trail({"ts_code": "600000.SH"}, {"optout_cooldown_tdays": 10},
|
||
set(), {("600000.SH", adv.R_TRAIL)}, True, out, None)
|
||
assert r is True and len(out["skipped"]) == 1 and not out["attached"]
|
||
finally:
|
||
adv._entry_tight_trail = s_e
|
||
|
||
|
||
@case("H 参数·紧止盈两键登记")
|
||
def _():
|
||
assert abs(ps.RUNTIME_EXTRA["PMS_TECH_TIGHT_TRAIL_GIVEBACK"][0] - 0.03) < 1e-9
|
||
assert abs(ps.RUNTIME_EXTRA["PMS_TECH_TIGHT_TRAIL_TARGET"][0] - 0.08) < 1e-9
|
||
|
||
|
||
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())
|