293 lines
13 KiB
Python
293 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""建议档位与真下股数对齐（2026-09-10）。全部离线，不连库不连 Redis 不调模型。
|
||
|
||
## 这一批在钉什么
|
||
|
||
2026-09-10 实测：155 上 8 张待确认提议，7 张卡上写着「建议方案：试探仓（目标 1.0%，
|
||
分批 100%）」，而每一张采纳后实际会买 3.00%（标准仓 6% 的第一批），正好三倍：
|
||
|
||
002812.SZ 1200股×48.55 实占 3.00% 卡上写 1%
|
||
300604.SZ 200股×264.81 实占 3.00% 卡上写 1%
|
||
300638.SZ 3300股×18.11 实占 3.05% 卡上写 1%
|
||
…只有 300118.SZ（标准仓）是自洽的
|
||
|
||
根因是次序：候选产出时 advise() 的档位是**进过**数量的（params_for 把 target_pct 塞进
|
||
stock_target_default），但研判回来之后 apply_judge 还会再降一档，那一步只改档位文字，
|
||
数量已经算完了。apply_judge 的说明原文写着「不改数量, 数量交人」，而页面上采纳按钮
|
||
根本没有填数量的地方，这个「交人」无处兑现 —— 采纳时后端直接抄提议里的股数原样落单。
|
||
|
||
A 按建议档位重算：正常降档、总规模从硬数字反推、批次比例跟着换。
|
||
B 买不足一手：抬到一手；一手就超过机械档的不改、交回调用方，并说清原因。
|
||
C 不该改的不改：档位没变、缺读数、没有建议方案。
|
||
D 方向只能是变小：重算之后的金额不许超过原来那一笔。
|
||
E 上游标了硬风险的候选一律交人 —— 哪怕档位是 full。
|
||
F 持仓的目标仓位在建仓成交入账时记一次，且只记第一次。
|
||
|
||
跑法：随全套一起跑（scripts/run_tests.py）。单跑是
|
||
python3 scripts/test_batch26_units.py
|
||
A 到 E 组在开发机上就能跑（纯逻辑与静态扫描）；F 组要 import 仓储层，开发机没装
|
||
sqlalchemy，得在容器里跑：
|
||
docker compose run --rm --no-deps -v $PWD:/app pms-web python scripts/test_batch26_units.py
|
||
"""
|
||
import os
|
||
import sys
|
||
import traceback
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from app.services import advice_service as adv # noqa: E402
|
||
|
||
RESULTS = []
|
||
|
||
|
||
def case(name):
|
||
def deco(fn):
|
||
RESULTS.append((name, fn))
|
||
return fn
|
||
return deco
|
||
|
||
|
||
SCALE = 2_000_000.0 # 155 上 PMS_TOTAL_SCALE 的实际值
|
||
|
||
|
||
def _hn(price, target_pct=0.06):
|
||
"""造一份候选产出时的硬数字。target_amount 是整只目标金额，与 target_pct 配套 ——
|
||
重算时的总规模就是从这两个数反推的，不另外去读参数表。"""
|
||
return {"price": price, "target_pct": target_pct,
|
||
"target_amount": round(SCALE * target_pct, 2)}
|
||
|
||
|
||
def _adv(tier, target_pct, splits, mech_pct=0.06):
|
||
return {"tier": tier, "target_pct": target_pct, "splits": tuple(splits),
|
||
"mechanical": {"target_pct": mech_pct, "splits": (0.5, 0.25, 0.25)}}
|
||
|
||
|
||
# ============================================================ A 按建议档位重算
|
||
@case("A1 试探仓降档：6% 的第一批 → 1% 一次买满（恩捷股份那张卡的真实读数）")
|
||
def _():
|
||
hn = _hn(48.55)
|
||
fix, why = adv.resize_to_advice(_adv("试探仓", 0.01, (1.0,)), hn, "002812.SZ")
|
||
assert fix, why
|
||
# 1% = 20000 元，48.55 一股，整百取整 → 400 股
|
||
assert fix["qty"] == 400, fix
|
||
assert abs(fix["target_pct"] - 0.01) < 1e-9, fix
|
||
assert abs(fix["base_amount"] - 400 * 48.55) < 0.01, fix
|
||
# 重算前是 1200 股（58,260 元），重算后 400 股（19,420 元），正好是原来的三分之一
|
||
assert fix["qty"] * 3 == 1200, fix
|
||
|
||
|
||
@case("A2 总规模从硬数字反推，不去读参数表（换一个规模照样对）")
|
||
def _():
|
||
hn = {"price": 10.0, "target_pct": 0.06, "target_amount": 60000.0} # 规模 100 万
|
||
fix, why = adv.resize_to_advice(_adv("试探仓", 0.01, (1.0,)), hn, "600000.SH")
|
||
assert fix, why
|
||
assert fix["qty"] == 1000, fix # 100 万 × 1% = 1 万元 ÷ 10 元 = 1000 股
|
||
|
||
|
||
@case("A3 批次比例跟着档位换（试探仓是一次买满，不许被降档梯子改成分批）")
|
||
def _():
|
||
fix, why = adv.resize_to_advice(_adv("试探仓", 0.01, (1.0,)), _hn(20.0), "600000.SH")
|
||
assert fix, why
|
||
assert fix["batch_scheme"] == "1.0", fix
|
||
# 一次买满：这一批的金额就等于整只目标金额
|
||
assert abs(fix["base_amount"] - fix["target_amount"]) < 0.01, fix
|
||
|
||
|
||
@case("A4 减半仓走两批，第一批只拿一半")
|
||
def _():
|
||
fix, why = adv.resize_to_advice(_adv("减半仓", 0.03, (0.5, 0.5)), _hn(20.0), "600000.SH")
|
||
assert fix, why
|
||
# 3% = 60000 元，两批各 30000，30000/20 = 1500 股
|
||
assert fix["qty"] == 1500, fix
|
||
assert fix["batch_scheme"] == "0.5,0.5", fix
|
||
|
||
|
||
# ============================================================ B 买不足一手
|
||
@case("B1 高价票买不足一手 → 抬到一手（300604.SZ 现价 264.81 的真实情形）")
|
||
def _():
|
||
fix, why = adv.resize_to_advice(_adv("试探仓", 0.01, (1.0,)), _hn(264.81), "300604.SZ")
|
||
assert fix, why
|
||
# 1% = 20000 元 ÷ 264.81 = 75 股，不足一手 → 抬到 100 股
|
||
assert fix["qty"] == 100, fix
|
||
assert "买不足一手" in (fix.get("note") or ""), fix
|
||
# 抬上去之后实际占比 1.32%，仍远小于机械档 6%
|
||
assert 0.013 < fix["target_pct"] < 0.014, fix
|
||
|
||
|
||
@case("B2 科创板一手是 200 股，不是 100 股")
|
||
def _():
|
||
fix, why = adv.resize_to_advice(_adv("试探仓", 0.01, (1.0,)), _hn(150.0), "688001.SH")
|
||
assert fix, why
|
||
# 1% = 20000 元 ÷ 150 = 133 股 < 200 股 → 抬到 200 股
|
||
assert fix["qty"] == 200, fix
|
||
|
||
|
||
@case("B3 一手就超过机械档 → 不改数量，交回调用方并说清原因")
|
||
def _():
|
||
# 一股 1500 元，一手 15 万；机械档 6% 是 12 万 —— 这只票对当前规模本来就太贵
|
||
fix, why = adv.resize_to_advice(_adv("试探仓", 0.01, (1.0,)), _hn(1500.0), "600000.SH")
|
||
assert fix is None, fix
|
||
assert "太贵" in why and "一手" in why, why
|
||
|
||
|
||
# ============================================================ C 不该改的不改
|
||
@case("C1 档位没变就不重算（标准仓那张卡本来就是自洽的）")
|
||
def _():
|
||
fix, why = adv.resize_to_advice(_adv("标准仓", 0.06, (0.5, 0.25, 0.25)), _hn(9.27), "300118.SZ")
|
||
assert fix is None, fix
|
||
assert "档位没变" in why, why
|
||
|
||
|
||
@case("C2 缺现价、缺目标金额、缺目标仓位，一律不改")
|
||
def _():
|
||
a = _adv("试探仓", 0.01, (1.0,))
|
||
for bad in ({"price": 0, "target_pct": 0.06, "target_amount": 120000.0},
|
||
{"price": 48.55, "target_pct": 0, "target_amount": 120000.0},
|
||
{"price": 48.55, "target_pct": 0.06, "target_amount": 0}):
|
||
fix, why = adv.resize_to_advice(a, bad, "002812.SZ")
|
||
assert fix is None, (bad, fix)
|
||
assert "缺少重算所需的读数" in why, why
|
||
|
||
|
||
@case("C3 没有建议方案或没有硬数字，不改也不抛")
|
||
def _():
|
||
assert adv.resize_to_advice(None, _hn(48.55), "002812.SZ")[0] is None
|
||
assert adv.resize_to_advice(_adv("试探仓", 0.01, (1.0,)), None, "002812.SZ")[0] is None
|
||
|
||
|
||
# ============================================================ D 方向只能是变小
|
||
@case("D1 重算之后这一笔的金额一定不超过原来那一笔（这条改动只许让钱变少）")
|
||
def _():
|
||
for price in (5.0, 9.27, 18.11, 25.24, 37.37, 48.55, 100.0, 264.81):
|
||
hn = _hn(price)
|
||
fix, _why = adv.resize_to_advice(_adv("试探仓", 0.01, (1.0,)), hn, "600000.SH")
|
||
if not fix:
|
||
continue
|
||
# 原来那一笔：6% 三批的第一批 50%
|
||
old_base = SCALE * 0.06 * 0.5
|
||
assert fix["base_amount"] <= old_base + 0.01, (price, fix["base_amount"], old_base)
|
||
|
||
|
||
@case("D2 抬到一手也不许超过机械档的整只目标金额")
|
||
def _():
|
||
for price in (200.0, 264.81, 500.0, 1000.0, 1199.0):
|
||
fix, _why = adv.resize_to_advice(_adv("试探仓", 0.01, (1.0,)), _hn(price), "600000.SH")
|
||
if fix:
|
||
assert fix["base_amount"] <= SCALE * 0.06 + 0.01, (price, fix)
|
||
|
||
|
||
# ============================================================ E 硬风险一律交人
|
||
@case("E1 上游标了硬风险的候选强制入人工队列，哪怕档位是 full")
|
||
def _():
|
||
src = open(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||
"app", "services", "proposal_service.py"), encoding="utf-8").read()
|
||
# 一票否决那一行必须把 risk_why 算进去。它排在档位判断之前，
|
||
# 所以 full 档也拦得住 —— 这正是 2026-09-10 补的那个洞。
|
||
line = [l for l in src.split("\n") if "force_queue = (" in l]
|
||
assert line, "找不到 force_queue 那一行，这道检查的前提没了"
|
||
seg = src[src.index("force_queue = ("):src.index("auto_exec = (not force_queue)")]
|
||
assert "risk_why" in seg, "硬风险没有算进一票否决：full 档下带风险的票会被自动买入"
|
||
# 交人原因要能说出是风险，不能笼统说「档位 propose_only」。
|
||
# 次序上重问放行仍排在最前（2026-09-09 拍板），硬风险紧随其后 —— 两者实际上
|
||
# 几乎不会同时出现：上游标了硬风险的票判决多半不是「候选」，走不到重问那一步。
|
||
assert "risk_why or c.get(" in src or "or risk_why or" in src, \
|
||
"交人原因里没有硬风险这一档"
|
||
|
||
|
||
@case("E2 硬风险的判据取的是候选硬数字里的 risk，与那条更窄的旁路同一个键")
|
||
def _():
|
||
src = open(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||
"app", "services", "proposal_service.py"), encoding="utf-8").read()
|
||
assert '_hn_risk = (c.get("hard_numbers") or {}).get("risk")' in src, \
|
||
"判据的键名与 _verdict_auto_exec_why 里那处不一致，两边会判出不同结果"
|
||
|
||
|
||
# ============================================================ F 持仓目标仓位
|
||
@case("F1 建仓成交入账时记一次目标仓位；已经有值就不覆盖")
|
||
def _():
|
||
from app.services import ledger_service as ls
|
||
from app.repo import pms_repo
|
||
|
||
calls = []
|
||
saved = (pms_repo.get_position, pms_repo.get_instruction,
|
||
pms_repo.get_proposal, pms_repo.update_position)
|
||
try:
|
||
pms_repo.get_instruction = lambda i: {"origin_type": "proposal", "origin_id": "PRP_X"}
|
||
pms_repo.get_proposal = lambda p: {"hard_numbers": {"advice": {"target_pct": 0.01, "tier": "试探仓"}}}
|
||
pms_repo.update_position = lambda code, **kw: calls.append((code, kw))
|
||
|
||
# 第一次：持仓行上还没有目标仓位 → 记
|
||
pms_repo.get_position = lambda c: {"target_pct": None}
|
||
ls._remember_target_pct("002812.SZ", "INS_1")
|
||
assert calls == [("002812.SZ", {"target_pct": 0.01})], calls
|
||
|
||
# 第二次：已经有值 → 不覆盖（目标仓位是建这只仓时定的意思，后续加仓不该改）
|
||
calls.clear()
|
||
pms_repo.get_position = lambda c: {"target_pct": 0.01}
|
||
ls._remember_target_pct("002812.SZ", "INS_2")
|
||
assert calls == [], calls
|
||
finally:
|
||
(pms_repo.get_position, pms_repo.get_instruction,
|
||
pms_repo.get_proposal, pms_repo.update_position) = saved
|
||
|
||
|
||
@case("F2 不是提议来源的指令不记；记不上也绝不能拦住成交入账")
|
||
def _():
|
||
from app.services import ledger_service as ls
|
||
from app.repo import pms_repo
|
||
|
||
calls = []
|
||
saved = (pms_repo.get_position, pms_repo.get_instruction,
|
||
pms_repo.get_proposal, pms_repo.update_position)
|
||
try:
|
||
pms_repo.get_position = lambda c: {"target_pct": None}
|
||
pms_repo.update_position = lambda code, **kw: calls.append((code, kw))
|
||
|
||
# 命令来源的指令：不记
|
||
pms_repo.get_instruction = lambda i: {"origin_type": "command", "origin_id": "CMD_1"}
|
||
pms_repo.get_proposal = lambda p: {}
|
||
ls._remember_target_pct("002812.SZ", "INS_1")
|
||
assert calls == [], calls
|
||
|
||
# 取数抛异常：吞掉，不往上冒（成交入账不能被这件事拦住）
|
||
def _boom(*a, **k):
|
||
raise RuntimeError("库挂了")
|
||
pms_repo.get_instruction = _boom
|
||
ls._remember_target_pct("002812.SZ", "INS_2") # 不抛就算过
|
||
assert calls == [], calls
|
||
|
||
# 没有指令号（外部成交并入 BASE 那种）：直接返回
|
||
ls._remember_target_pct("002812.SZ", None)
|
||
assert calls == [], calls
|
||
finally:
|
||
(pms_repo.get_position, pms_repo.get_instruction,
|
||
pms_repo.get_proposal, pms_repo.update_position) = saved
|
||
|
||
|
||
@case("F3 目标仓位这一列在仓储层的可写白名单里（不在的话上面那次写会被静默拒掉）")
|
||
def _():
|
||
from app.repo import pms_repo
|
||
assert "target_pct" in pms_repo.POSITION_COLS
|
||
|
||
|
||
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())
|