From b3fdb4a4e6520c073f15d10d79b3e710b4686478 Mon Sep 17 00:00:00 2001 From: zlt Date: Mon, 27 Jul 2026 17:12:09 +0800 Subject: [PATCH] =?UTF-8?q?=E7=AC=AC=E4=BA=8C=E6=89=B9:=20=E5=91=BD?= =?UTF-8?q?=E4=BB=A4=E7=8A=B6=E6=80=81=E6=9C=BA/=E6=96=B9=E6=A1=88?= =?UTF-8?q?=E7=94=9F=E6=88=90=E5=99=A8/=E5=9B=9E=E6=94=BE=E5=AF=B9?= =?UTF-8?q?=E8=B4=A6/=E7=AE=A1=E7=90=86=E9=A1=B5=E9=9D=A2/=E8=B0=83?= =?UTF-8?q?=E5=BA=A6=E5=99=A8=E9=AA=A8=E6=9E=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 3 + README.md | 82 +++- app/__init__.py | 0 app/core/__init__.py | 0 app/core/command_spec.py | 440 +++++++++++++++++++ app/core/planner.py | 552 ++++++++++++++++++++++++ app/core/recon.py | 321 ++++++++++++++ app/core/tradedays.py | 104 +++++ app/db/__init__.py | 0 app/db/session.py | 191 +++++++++ app/repo/__init__.py | 0 app/repo/downstream_repo.py | 201 +++++++++ app/repo/pms_repo.py | 521 +++++++++++++++++++++++ app/scheduler.py | 185 ++++++++ app/services/__init__.py | 0 app/services/command_service.py | 494 ++++++++++++++++++++++ app/services/industry.py | 88 ++++ app/services/ledger_service.py | 474 +++++++++++++++++++++ app/services/market.py | 194 +++++++++ app/services/param_store.py | 273 ++++++++++++ app/services/portfolio.py | 156 +++++++ app/web/__init__.py | 0 app/web/main.py | 321 +++++++++++++- app/web/static/index.html | 596 ++++++++++++++++++++++++++ config/settings.py | 2 + scripts/check_db.py | 125 ++++++ scripts/run_tests.py | 41 ++ scripts/test_batch2_units.py | 551 ++++++++++++++++++++++++ scripts/test_wiring.py | 725 ++++++++++++++++++++++++++++++++ 29 files changed, 6611 insertions(+), 29 deletions(-) create mode 100644 app/__init__.py create mode 100644 app/core/__init__.py create mode 100644 app/core/command_spec.py create mode 100644 app/core/planner.py create mode 100644 app/core/recon.py create mode 100644 app/core/tradedays.py create mode 100644 app/db/__init__.py create mode 100644 app/db/session.py create mode 100644 app/repo/__init__.py create mode 100644 app/repo/downstream_repo.py create mode 100644 app/repo/pms_repo.py create mode 100644 app/scheduler.py create mode 100644 app/services/__init__.py create mode 100644 app/services/command_service.py create mode 100644 app/services/industry.py create mode 100644 app/services/ledger_service.py create mode 100644 app/services/market.py create mode 100644 app/services/param_store.py create mode 100644 app/services/portfolio.py create mode 100644 app/web/__init__.py create mode 100644 app/web/static/index.html create mode 100644 scripts/check_db.py create mode 100644 scripts/run_tests.py create mode 100644 scripts/test_batch2_units.py create mode 100644 scripts/test_wiring.py diff --git a/.gitignore b/.gitignore index 2555c47..e23a8e0 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,6 @@ logs/ # PyCharm 工程文件目前已跟踪; 如决定不入库, 取消下行注释并执行: # git rm -r --cached .idea && git commit -m "untrack .idea" #.idea/ + +# 本地临时: Cowork 会话搬运产物, 可整目录删除 +_to_delete/ diff --git a/README.md b/README.md index 5dfd48c..5c0d168 100644 --- a/README.md +++ b/README.md @@ -14,8 +14,36 @@ | `QMT_INTERFACE_REQUIREMENTS.md` | 与 QMT 侧下游系统协商用的数据与接口需求清单（含资金快照、统一指令通道建议 DDL），按编号答复回填 | | `ddl_pms_v1.sql` | PMS 全部自有表建表语句（153 代理侧，10 张） | | `config/settings.py` | 配置（基础设施键名对齐 bionic；业务参数为初值，页面调参持久化到 `pms_runtime_param` 后优先） | -| `app/core/sizer.py` / `app/core/cushion.py` | 首批核心纯逻辑：批次拆分与一手合并、组合约束、风险披露；摊薄成本、安全垫状态机、保垫触发、卖出核销次序 | -| `scripts/test_core_units.py` | 上述模块的实机单测（`python scripts/test_core_units.py`，零外部依赖，14 例） | + +## 模块地图 + +``` +app/ + core/ 纯逻辑, 零外部依赖, 可单测 —— 系统的算数与纪律都在这里 + sizer.py 批次拆分与一手合并 / 组合约束 / 风险敞口披露 + cushion.py 摊薄成本 / 安全垫状态机 / 保垫触发 / 卖出核销次序 + command_spec.py 命令目录(A/B/C 全量 27 类) / 参数校验 / 双状态机 / 冲突识别 + planner.py 方案生成器: 降仓凑额四档 / 升仓 / 建仓 / 清仓 / 行业 / 撤单 + recon.py 成交认领与入账映射 / 对账差异与修正 / 除权检测 / T+1 可用量 + tradedays.py 交易日历: 调度守卫与执行窗口计算 + db/session.py 三库连接 + **严格单表访问守卫** (JOIN/逗号连表/跨表子查询一律拒绝) + repo/ 单表数据访问: pms_repo (自有 10 表) / downstream_repo (下游只读三表) + services/ 编排层 + param_store.py 运行参数中心 (表值优先于 settings 初值, 页面调参即时生效) + portfolio.py 组合快照 (账本+行情+行业 → 方案/规则闸/页面的统一输入) + command_service.py 命令下达→校验→冲突→生效/规划→进度推进→撤销 + ledger_service.py 成交回放 / 对账 / 除权 / 盘前 / 日终结算 / 运营日报 + market.py 行情 (Redis db13) 与参考位 (决策系统主口径 + 兜底自算) + industry.py 行业划分可插拔适配器 (custom_table / gp_stock_category / 停用) + web/ FastAPI + 单页 (Vue3 + ElementPlus)，页面四块 + 运维/日报抽屉 + scheduler.py Celery beat 调度总表 (设计 §10 八个调度位 + 三条守卫) +scripts/ + run_tests.py 一次跑完全部单测 (见下方「Docker 部署」) + test_core_units.py 仓位与安全垫核心逻辑 14 例 + test_batch2_units.py 命令 / 方案 / 回放对账 纯逻辑 35 例 + test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) 18 例 + check_db.py 实机连通性与表结构自检 (需真实 .env) +``` ## 三条铁律 @@ -25,29 +53,59 @@ ## Docker 部署（项目统一以容器方式构建运行） -服务共用一个镜像：`pms-web`（管理页面，端口 38100）+ `pms-beat` / `pms-worker`（Celery 调度与执行，挂在 `sched` profile 下，调度器代码就绪后启用）。 +服务共用一个镜像：`pms-web`（管理页面，端口 38100）+ `pms-beat` / `pms-worker`（Celery 调度与执行，挂在 `sched` profile 下）。 ```bash # 服务器首次部署 git clone <仓库地址> && cd tradingSystem cp .env.example .env && vim .env # 填入真实连接串 (.env 不入库) docker compose build # 默认走清华 PyPI 镜像; 可 --build-arg PIP_INDEX_URL=... 覆盖 -docker compose run --rm pms-web python scripts/test_core_units.py # 构建验证: 应输出 ALL PASS (14 cases) -docker compose up -d # 当前仅启动 pms-web -curl http://127.0.0.1:38100/health # 健康检查 + 配置装载自证 -# 调度器代码就绪后 -docker compose --profile sched up -d +# 构建验证 (不连库, 秒级): 应输出 ALL SUITES PASS +docker compose run --rm pms-web python scripts/run_tests.py +# 实机自检 (连库, 需 .env): 三库连通 + pms_* 十表 + 下游表结构 + 行情 Redis +docker compose run --rm pms-web python scripts/check_db.py + +docker compose up -d # 管理页面 +curl http://127.0.0.1:38100/health # 健康检查 + 配置装载自证 + 库连通自证 +# 浏览器打开 http://<服务器IP>:38100/ → 参数设置 / 命令台 / 持仓与账本 / 提议确认 + +docker compose --profile sched up -d # 启用调度器 (beat + worker) +docker compose logs -f pms-beat pms-worker # 日常更新 git pull && docker compose build && docker compose up -d ``` -基础镜像 `python:3.11-slim` 拉取慢时，先给服务器 Docker 配置 registry 镜像加速。日志落 `./logs`（已挂载卷）；容器时区 Asia/Shanghai。 +基础镜像 `python:3.11-slim` 拉取慢时，先给服务器 Docker 配置 registry 镜像加速。日志落 `./logs`（已挂载卷）；容器时区 Asia/Shanghai。管理页面的前端资源（Vue3 / ElementPlus / axios）走 unpkg CDN，浏览器需能访问外网；若内网隔离，把页面头部三行 ` + + + + + +
+ +
+

PMS 持仓管理系统

+ v{{ health.version || '-' }} · {{ health.now || '' }} + 交易日 + 非交易日 + +
+ + 买入 {{ ov.buy_halt ? '已暂停':'正常' }} + + 执行 {{ ov.exec_halt ? '已暂停':'正常' }} + 自主档位 {{ ov.autonomy || '-' }} + 日报 + 运维 + 刷新 +
+ +
+ + + + diff --git a/config/settings.py b/config/settings.py index f70ea18..a59bff2 100644 --- a/config/settings.py +++ b/config/settings.py @@ -65,6 +65,8 @@ class Settings(BaseSettings): PMS_NO_CHASE_MA5: float = 0.06 # 距 MA5 超此幅度不追买 PMS_BUILD_WINDOW_TDAYS: int = 10 # 建仓期窗口 (交易日) PMS_FILL_MAX_LOSS: float = -0.03 # 浮亏深于此不走回踩补足 (转补仓域) + PMS_WEAK_NEG_DAYS: int = 5 # 降仓「清弱票」判定: 安全垫连续为负 N 日 (设计 §3.2) + PMS_PROPOSAL_TTL_HOURS: int = 24 # 自主提议待确认有效期 (超时置 EXPIRED) # --- 行业约束 (硬拦截; 数据源接口化) --- PMS_SECTOR_SOURCE: str = "" # "" = 停用并页面提示 / custom_table / gp_stock_category diff --git a/scripts/check_db.py b/scripts/check_db.py new file mode 100644 index 0000000..f8073c2 --- /dev/null +++ b/scripts/check_db.py @@ -0,0 +1,125 @@ +# -*- coding: utf-8 -*- +""" +连通性与表结构自检 (实机, 需 .env 真实连接串) +============================================== +运行: docker compose run --rm pms-web python scripts/check_db.py + +检查项: + 1. 三个库连通性 (153 代理 / 因子库 / 指数库) + 2. pms_* 十张表是否存在与当前行数 + 3. 下游三表可读性 + 持仓数量列探测结果 (回填 QMT_INTERFACE_REQUIREMENTS A1/D1 用) + 4. 行情 Redis (db13) 连通性与样本键 + 5. 运行参数表当前生效值 +全部通过退出码 0; 任一 FAIL 退出码 1 (WARN 不影响退出码)。 +""" +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +PMS_TABLES = ["pms_command", "pms_plan", "pms_position", "pms_lot", "pms_instruction", + "pms_proposal", "pms_action_ledger", "pms_daily_report", "pms_industry_map", + "pms_runtime_param"] +DOWNSTREAM = ["trading_position", "trading_order", "trading_buy_plan"] + +FAILED, WARNED = [], [] + + +def line(tag, msg): + print(f" {tag:<6}{msg}") + + +def fail(msg): + FAILED.append(msg) + line("FAIL", msg) + + +def warn(msg): + WARNED.append(msg) + line("WARN", msg) + + +def main(): + from app.db import session as dbs + + print("\n[1] 数据库连通性") + for name, label in (("proxy", "153 代理 (pms_* / trading_*)"), + ("factor", "因子分表库"), ("index", "大盘指数库")): + r = dbs.ping(name) + if r["ok"]: + line("OK", f"{label}") + elif name == "proxy": + fail(f"{label} 连接失败: {r['error']}") + else: + warn(f"{label} 连接失败 (非致命): {r['error']}") + + print("\n[2] PMS 自有表") + for t in PMS_TABLES: + try: + r = dbs.fetch_one(f"SELECT COUNT(*) AS n FROM {t}") + line("OK", f"{t:<20} {r['n']} 行") + except Exception as e: + fail(f"{t} 不可用: {type(e).__name__}: {e}") + + print("\n[3] 下游只读表") + from app.repo import downstream_repo + for t in DOWNSTREAM: + try: + cols = downstream_repo.describe(t) + line("OK", f"{t:<20} {len(cols)} 列: " + f"{', '.join(str(c.get('Field')) for c in cols[:10])}" + f"{' ...' if len(cols) > 10 else ''}") + except Exception as e: + warn(f"{t} 读取失败: {type(e).__name__}: {e}") + try: + ds = downstream_repo.fetch_positions() + line("OK" if ds["columns"].get("qty") else "WARN", + f"持仓列探测: {ds['columns']} (共 {ds['raw_count']} 行)") + if ds["raw_count"] and not ds["columns"].get("qty"): + warn("未识别出持仓数量列 —— 请按 QMT 需求清单 A1/D1 取得 DDL 后补 QTY_CANDIDATES") + except Exception as e: + warn(f"持仓快照读取失败: {e}") + + print("\n[4] 行情 Redis (db13)") + try: + from app.services import market + pos = [] + try: + from app.repo import pms_repo + pos = [p["ts_code"] for p in pms_repo.list_positions(only_open=True)][:3] + except Exception: + pass + probe = pos or ["600000.SH"] + got = {c: market.get_price(c) for c in probe} + if any(v for v in got.values()): + line("OK", f"实时价样本: {got}") + else: + warn(f"取不到实时价样本 {got} (非交易时段属正常)") + except Exception as e: + warn(f"Redis 行情不可用: {type(e).__name__}: {e}") + + print("\n[5] 运行参数 (表值优先于 settings 初值)") + try: + from app.services import param_store + snap = param_store.snapshot() + from_table = [p for p in snap["params"] if p["source"] == "table"] + line("OK", f"可调参数 {len(snap['params'])} 项, 其中页面已改写 {len(from_table)} 项") + for k in ("PMS_TOTAL_SCALE", "PMS_PORTFOLIO_CAP", "PMS_STOCK_CAP", "PMS_AUTONOMY", + "PMS_SECTOR_SOURCE"): + line("", f"{k:<26} = {param_store.get(k)}") + if snap.get("source_error"): + warn(f"参数表读取异常: {snap['source_error']}") + except Exception as e: + fail(f"参数中心不可用: {type(e).__name__}: {e}") + + print("\n" + "-" * 62) + if FAILED: + print(f"FAILED: {len(FAILED)} 项致命问题, {len(WARNED)} 项告警") + for m in FAILED: + print(" - " + m) + sys.exit(1) + print(f"ALL OK ({len(WARNED)} 项告警)" if WARNED else "ALL OK") + + +if __name__ == "__main__": + main() diff --git a/scripts/run_tests.py b/scripts/run_tests.py new file mode 100644 index 0000000..193eeed --- /dev/null +++ b/scripts/run_tests.py @@ -0,0 +1,41 @@ +# -*- coding: utf-8 -*- +""" +全部单测一次跑完 (零外部依赖, 不连库) +====================================== +运行: docker compose run --rm pms-web python scripts/run_tests.py + +包含: + test_core_units.py 仓位规划器 / 安全垫与成本账 (14 例) + test_batch2_units.py 命令状态机 / 方案生成器 / 回放对账纯逻辑 (35 例) + test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (18 例) +任一子集失败即整体失败 (退出码 1)。 +""" +import os +import subprocess +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +ROOT = os.path.dirname(HERE) +SUITES = ["test_core_units.py", "test_batch2_units.py", "test_wiring.py"] + + +def main(): + failed = [] + for s in SUITES: + path = os.path.join(HERE, s) + if not os.path.exists(path): + print(f"== 跳过 {s} (文件不存在)") + continue + print(f"\n=== {s} " + "=" * (52 - len(s))) + r = subprocess.run([sys.executable, path], cwd=ROOT) + if r.returncode != 0: + failed.append(s) + print("\n" + "=" * 62) + if failed: + print(f"SUITE FAILED: {', '.join(failed)}") + sys.exit(1) + print("ALL SUITES PASS") + + +if __name__ == "__main__": + main() diff --git a/scripts/test_batch2_units.py b/scripts/test_batch2_units.py new file mode 100644 index 0000000..895b304 --- /dev/null +++ b/scripts/test_batch2_units.py @@ -0,0 +1,551 @@ +# -*- coding: utf-8 -*- +""" +第二批模块单测 (实机运行, 零外部依赖) +====================================== +运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch2_units.py +覆盖: command_spec 命令目录/校验/状态机/冲突; planner 降仓凑额四档+升仓+建仓+个股方案; + recon 回放认领/外部成交/对账差异/除权/T+1可用量。 +约定同 test_core_units.py: 全过输出 "ALL PASS (n cases)" 退出码 0, 任一失败退出码 1。 +""" +import os +import sys +import traceback + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from app.core import command_spec as cs # noqa: E402 +from app.core import planner as pl # noqa: E402 +from app.core import recon as rc # noqa: E402 + +RESULTS = [] + + +def case(name): + def deco(fn): + RESULTS.append((name, fn)) + return fn + return deco + + +# ================================================================ command_spec +@case("命令校验·降仓参数归一 (10% 字符串 → 0.1, 窗口取默认值)") +def _(): + p, e = cs.validate("REDUCE_EXPOSURE", {"pct": "10%"}) + assert e == [], e + assert abs(p["pct"] - 0.10) < 1e-9 and p["window_tdays"] == 3, p + p2, e2 = cs.validate("REDUCE_EXPOSURE", {"pct": 0.1, "window_tdays": 5}) + assert e2 == [] and p2["window_tdays"] == 5, (p2, e2) + + +@case("命令校验·缺参/越界/未知字段/未知命令全部拒绝") +def _(): + _, e1 = cs.validate("REDUCE_EXPOSURE", {}) + assert any(x.startswith("MISSING") for x in e1), e1 + _, e2 = cs.validate("SET_PORTFOLIO_CAP", {"cap": 1.5}) + assert any(x.startswith("OUT_OF_RANGE") for x in e2), e2 + _, e3 = cs.validate("SET_MAX_NAMES", {"n": 15, "foo": 1}) + assert any(x.startswith("UNKNOWN_FIELD") for x in e3), e3 + _, e4 = cs.validate("NOT_A_CMD", {}) + assert any(x.startswith("UNKNOWN_CMD") for x in e4), e4 + _, e5 = cs.validate("SET_AUTONOMY", {"mode": "auto"}) + assert any(x.startswith("BAD_ENUM") for x in e5), e5 + _, e6 = cs.validate("LIQUIDATE_ALL", {}) # 紧急清仓必须二次确认 + assert any(x.startswith("MISSING") for x in e6), e6 + + +@case("命令校验·股票代码归一 (点式/前缀式/裸码)") +def _(): + n = cs.normalize_code + assert n("600000.SH") == "600000.SH" + assert n("sh600000") == "600000.SH" + assert n("600000") == "600000.SH" + assert n("000001") == "000001.SZ" + assert n("300750") == "300750.SZ" + assert n("430047") == "430047.BJ" + p, e = cs.validate("EXIT_STOCK", {"ts_code": "sz000001"}) + assert e == [] and p["ts_code"] == "000001.SZ", (p, e) + + +@case("状态机·任务命令合法链路与非法迁移") +def _(): + st = cs.ST_PENDING + for nxt in (cs.ST_PLANNING, cs.ST_EXECUTING, cs.ST_DONE): + st = cs.transition(cs.CLS_TASK, st, nxt) + assert st == cs.ST_DONE + assert not cs.can_transition(cs.CLS_TASK, cs.ST_PENDING, cs.ST_DONE) + assert not cs.can_transition(cs.CLS_TASK, cs.ST_DONE, cs.ST_EXECUTING) + try: + cs.transition(cs.CLS_TASK, cs.ST_DONE, cs.ST_EXECUTING) + assert False, "非法迁移未拦截" + except ValueError: + pass + # 部分完成可顺延回执行中 + assert cs.can_transition(cs.CLS_TASK, cs.ST_PARTIAL, cs.ST_EXECUTING) + + +@case("状态机·参数命令生效与被覆盖") +def _(): + assert cs.can_transition(cs.CLS_PARAM, cs.ST_PENDING, cs.ST_EFFECTIVE) + assert cs.can_transition(cs.CLS_PARAM, cs.ST_EFFECTIVE, cs.ST_SUPERSEDED) + assert not cs.can_transition(cs.CLS_PARAM, cs.ST_SUPERSEDED, cs.ST_EFFECTIVE) + + +@case("状态机·任务进度结算 (达标DONE / 窗口末未达标PARTIAL / 途中EXECUTING)") +def _(): + assert cs.settle_task_status(200_000, 200_000, False) == cs.ST_DONE + assert cs.settle_task_status(200_000, 120_000, True) == cs.ST_PARTIAL + assert cs.settle_task_status(200_000, 120_000, False) == cs.ST_EXECUTING + assert cs.settle_task_status(0, 0, False) == cs.ST_DONE + + +@case("命令冲突·组合反向/同股反向/异股不冲突/重复下达") +def _(): + active = [{"command_id": "CMD_1", "cmd_type": "INCREASE_EXPOSURE"}, + {"command_id": "CMD_2", "cmd_type": "OPEN_TARGET", "ts_code": "600000.SH"}] + c1 = cs.detect_conflicts({"cmd_type": "REDUCE_EXPOSURE"}, active) + assert len(c1) == 1 and c1[0]["with_command_id"] == "CMD_1", c1 + c2 = cs.detect_conflicts({"cmd_type": "EXIT_STOCK", "ts_code": "600000.SH"}, active) + assert len(c2) == 1 and c2[0]["with_cmd_type"] == "OPEN_TARGET", c2 + c3 = cs.detect_conflicts({"cmd_type": "EXIT_STOCK", "ts_code": "000001.SZ"}, active) + assert c3 == [], c3 + c4 = cs.detect_conflicts({"cmd_type": "OPEN_TARGET", "ts_code": "600000.SH"}, active) + assert any("重复下达" in x["reason"] for x in c4), c4 + # 冻结中的票不得建仓 + c5 = cs.detect_conflicts({"cmd_type": "OPEN_TARGET", "ts_code": "600519.SH"}, + [{"command_id": "CMD_9", "cmd_type": "FREEZE_STOCK", + "ts_code": "600519.SH"}]) + assert len(c5) == 1 and "冻结" in c5[0]["reason"], c5 + + +@case("命令目录·完整性与行业数据源未配置时置灰") +def _(): + all_cmds = cs.list_commands() + assert len(all_cmds) >= 25, len(all_cmds) + for c in all_cmds: + assert c["label"] and c["cls"] in (cs.CLS_PARAM, cs.CLS_TASK), c + spec = cs.SPECS[c["cmd_type"]] + if spec["cls"] == cs.CLS_TASK: + assert "planner" in spec, c["cmd_type"] + if spec["cls"] == cs.CLS_PARAM and spec["scope"] == "global": + assert "param_key" in spec and "value_field" in spec, c["cmd_type"] + off = {c["cmd_type"]: c for c in cs.list_commands(sector_source_ready=False)} + assert off["SECTOR_EXIT"]["disabled"] and off["SECTOR_CAP"]["disabled"] + assert not off["REDUCE_EXPOSURE"]["disabled"] + on = {c["cmd_type"]: c for c in cs.list_commands(sector_source_ready=True)} + assert not on["SECTOR_EXIT"]["disabled"] + assert {c["group"] for c in all_cmds} == {"A", "B", "C"} + + +# ================================================================ planner · 降仓 +def _positions_demo(): + """设计 §3.2 场景: 规模 200 万, 组合市值 42 万。""" + return [ + # A: 浮盈 12%, 三批齐全 (底仓 7000, 补足 3500, 加仓 3500) + dict(ts_code="600000.SH", price=10.0, total_qty=14000, base_qty=7000, + fill_qty=3500, add_qty=3500, dca_qty=0, t0_qty=0, + cushion_pct=0.12, cushion_state="SOLID", neg_cushion_days=0, + status="HOLDING", frozen_reason="NONE"), + # B: 弱票 —— 安全垫 -9% 且连续 6 日为负 + dict(ts_code="000001.SZ", price=8.0, total_qty=10000, base_qty=10000, + fill_qty=0, add_qty=0, dca_qty=0, t0_qty=0, + cushion_pct=-0.09, cushion_state="NONE", neg_cushion_days=6, + status="HOLDING", frozen_reason="NONE"), + # C: 微盈 2%, 只有底仓 (收利润档无可卖) + dict(ts_code="600519.SH", price=20.0, total_qty=5000, base_qty=5000, + fill_qty=0, add_qty=0, dca_qty=0, t0_qty=0, + cushion_pct=0.02, cushion_state="THIN", neg_cushion_days=0, + status="HOLDING", frozen_reason="NONE"), + # D: 浮盈 5%, 可卖 5000 股 (被命令冻结, 但冻结只禁增持不禁减持) + dict(ts_code="300750.SZ", price=5.0, total_qty=20000, base_qty=15000, + fill_qty=2500, add_qty=2500, dca_qty=0, t0_qty=0, + cushion_pct=0.05, cushion_state="SOLID", neg_cushion_days=0, + status="HOLDING", frozen_reason="COMMAND_HALT"), + ] + + +@case("降仓凑额·四档全走通 (释放20万: 撤单→清弱票8万→收利润9.5万→等比微减凑齐)") +def _(): + r = pl.plan_reduce_exposure( + release_amount=200_000, positions=_positions_demo(), + pending_buys=[{"instruction_id": "INS_X", "ts_code": "600036.SH", + "qty": 1000, "amount": 10_000}]) + tiers = [i["tier"] for i in r["items"]] + assert tiers[0] == "1_停新买", tiers + assert r["items"][0]["action"] == pl.A_HALT + assert r["items"][0]["cancel_instruction_id"] == "INS_X" + + weak = [i for i in r["items"] if i["tier"] == "2_清弱票"] + assert len(weak) == 1 and weak[0]["ts_code"] == "000001.SZ", weak + assert weak[0]["action"] == pl.A_EXIT and weak[0]["qty"] == 10000 + assert abs(weak[0]["amount"] - 80_000) < 1e-6 + + harvest = [i for i in r["items"] if i["tier"] == "3_收利润"] + # 垫子厚的先收: A(12%) 卖 7000 股 = 7万, 再 D(5%) 卖 5000 股 = 2.5万 + assert [i["ts_code"] for i in harvest] == ["600000.SH", "300750.SZ"], harvest + assert [i["qty"] for i in harvest] == [7000, 5000], harvest + assert all(i["action"] == pl.A_TRIM for i in harvest) + + prorata = [i for i in r["items"] if i["tier"] == "4_等比微减"] + # 缺口 2.5 万, 按剩余市值 C(10万)/D(7.5万)/A(7万) 权重摊派 + 取整补齐 + assert {i["ts_code"] for i in prorata} == {"600519.SH", "300750.SZ", "600000.SH"}, prorata + assert all(i["qty"] % 100 == 0 for i in prorata), prorata + assert r["ok"] and r["gap"] == 0.0, r + assert r["planned_amount"] >= 200_000, r + assert r["planned_amount"] == 201_500.0, r["planned_amount"] # 取整到手的必然轻微超额 + + +@case("降仓凑额·不超卖 (每票累计计划量 ≤ 持仓量)") +def _(): + r = pl.plan_reduce_exposure(release_amount=350_000, positions=_positions_demo()) + hold = {p["ts_code"]: p["total_qty"] for p in _positions_demo()} + used = {} + for i in r["items"]: + if i["action"] in (pl.A_EXIT, pl.A_TRIM): + used[i["ts_code"]] = used.get(i["ts_code"], 0) + i["qty"] + for c, q in used.items(): + assert q <= hold[c], (c, q, hold[c]) + + +@case("降仓凑额·可减持仓不足时给缺口并标不达标") +def _(): + r = pl.plan_reduce_exposure(release_amount=1_000_000, positions=_positions_demo()) + assert not r["ok"] and r["gap"] > 0, r + assert any("缺口" in n for n in r["notes"]), r["notes"] + assert r["planned_amount"] <= 420_000 + 1e-6, r + + +@case("降仓凑额·清弱票为整票清仓, 超目标时留痕说明") +def _(): + r = pl.plan_reduce_exposure(release_amount=30_000, positions=_positions_demo()) + weak = [i for i in r["items"] if i["tier"] == "2_清弱票"] + assert len(weak) == 1 and weak[0]["qty"] == 10000, weak # 不为凑额拆卖弱票 + assert any("超目标" in n for n in r["notes"]), r["notes"] + assert len([i for i in r["items"] if i["tier"] != "2_清弱票"]) == 0, r["items"] + + +@case("降仓凑额·跳过已有在途方案的票 + 零释放额只撤单") +def _(): + r = pl.plan_reduce_exposure(release_amount=100_000, positions=_positions_demo(), + exclude_codes=["000001.SZ"]) + assert all(i["ts_code"] != "000001.SZ" for i in r["items"]), r["items"] + r2 = pl.plan_reduce_exposure(release_amount=0, positions=_positions_demo(), + pending_buys=[{"instruction_id": "I1", "ts_code": "600000.SH", + "qty": 100, "amount": 1000}]) + assert len(r2["items"]) == 1 and r2["items"][0]["action"] == pl.A_HALT, r2 + assert not r2["ok"] + + +# ================================================================ planner · 建仓/升仓/个股 +def _ctx(**kw): + d = dict(scale=2_000_000, portfolio_cap=0.60, stock_cap=0.08, max_names=15, + portfolio_mv=800_000, names_count=5, stock_mv=0.0, is_new_name=True, + sector=None, sector_names=0, sector_mv=0.0, + sector_max_names=4, sector_max_ratio=0.40, cash_reserve=0.0) + d.update(kw) + return d + + +@case("建仓命令·50/25/25 分批 (12万@10元 → 6000/3000/3000, 后两批 gated 待引擎解锁)") +def _(): + r = pl.plan_open_target(ts_code="600000.SH", target_pct=0.06, price=10.0, ctx=_ctx()) + assert r["ok"], r + assert [i["qty"] for i in r["items"]] == [6000, 3000, 3000], r["items"] + assert [i["action"] for i in r["items"]] == [pl.A_OPEN, pl.A_FILL, pl.A_ADD] + assert [i["gated"] for i in r["items"]] == [False, True, True] + assert all(i["side"] == pl.SIDE_BUY for i in r["items"]) + assert abs(r["planned_amount"] - 120_000) < 1e-6 + + +@case("建仓命令·一手合并 (12万@700元 → 合并为单批 100 股并留痕)") +def _(): + r = pl.plan_open_target(ts_code="600519.SH", target_pct=0.06, price=700.0, ctx=_ctx()) + assert r["ok"] and len(r["items"]) == 1 and r["items"][0]["qty"] == 100, r + assert any("批次自动合并" in n for n in r["notes"]), r["notes"] + + +@case("建仓命令·单股上限硬拦截 + 已达目标不重复建仓") +def _(): + r = pl.plan_open_target(ts_code="600000.SH", target_pct=0.10, price=10.0, ctx=_ctx()) + assert not r["ok"] and r["items"] == [], r + assert any(x.startswith("STOCK_CAP") for x in r["rejects"][0]["reasons"]), r["rejects"] + r2 = pl.plan_open_target(ts_code="600000.SH", target_pct=0.06, price=10.0, + ctx=_ctx(stock_mv=130_000)) + assert r2["ok"] and r2["items"] == [], r2 + + +@case("建仓命令·预留现金约束 (与总仓上限双重约束)") +def _(): + ctx = _ctx(portfolio_mv=1_750_000, portfolio_cap=0.95, cash_reserve=0.10) + r = pl.plan_open_target(ts_code="600000.SH", target_pct=0.06, price=10.0, ctx=ctx) + assert not r["ok"], r + assert any(x.startswith("CASH_RESERVE") for x in r["rejects"][0]["reasons"]), r["rejects"] + + +@case("建仓命令·行业硬拦截 (同行业 4 只已满)") +def _(): + ctx = _ctx(sector_names_map={"半导体": 4}, sector_mv_map={"半导体": 300_000}) + r = pl.plan_increase_exposure( + add_amount=120_000, positions=[], ctx=ctx, + candidates=[{"ts_code": "688981.SH", "price": 20.0, "score": 0.9, "sector": "半导体"}], + params={"stock_target_default": 0.06}) + assert r["items"] == [] and r["rejects"], r + assert any(x.startswith("SECTOR_NAMES") for x in r["rejects"][0]["reasons"]), r["rejects"] + + +@case("升仓命令·既有垫厚票补到目标 + 候选池新票建仓, 累计口径校验上限") +def _(): + holdings = [dict(ts_code="600000.SH", price=10.0, total_qty=8000, base_qty=8000, + cushion_pct=0.08, cushion_state="SOLID", target_pct=0.06, + frozen_reason="NONE", sector=None)] + cands = [{"ts_code": "688981.SH", "price": 20.0, "score": 0.9, "sector": "半导体"}, + {"ts_code": "000651.SZ", "price": 15.0, "score": 0.8, "sector": "家电"}] + r = pl.plan_increase_exposure(add_amount=200_000, positions=holdings, candidates=cands, + ctx=_ctx(), params={"stock_target_default": 0.06}) + adds = [i for i in r["items"] if i["action"] == pl.A_ADD and i["tier"] == "1_补既有"] + assert len(adds) == 1 and adds[0]["ts_code"] == "600000.SH" and adds[0]["qty"] == 4000, adds + new_codes = {i["ts_code"] for i in r["items"] if i["tier"].startswith("批次_")} + assert new_codes == {"688981.SH", "000651.SZ"}, new_codes + assert r["ok"] and abs(r["planned_amount"] - 200_000) < 1e-6, r + + +@case("升仓命令·冻结票与非厚垫票不补仓") +def _(): + holdings = [dict(ts_code="600000.SH", price=10.0, total_qty=8000, base_qty=8000, + cushion_pct=0.08, cushion_state="SOLID", target_pct=0.06, + frozen_reason="COMMAND_HALT", sector=None), + dict(ts_code="000001.SZ", price=8.0, total_qty=5000, base_qty=5000, + cushion_pct=0.01, cushion_state="THIN", target_pct=0.06, + frozen_reason="NONE", sector=None)] + r = pl.plan_increase_exposure(add_amount=100_000, positions=holdings, candidates=[], + ctx=_ctx(), params={}) + assert r["items"] == [] and not r["ok"], r + + +@case("个股命令·清仓(含零股)/减至X%/减至0等价清仓") +def _(): + pos = dict(ts_code="600000.SH", price=10.0, total_qty=14050, base_qty=14050) + r = pl.plan_exit_stock(ts_code="600000.SH", position=pos) + assert r["items"][0]["qty"] == 14050, r # 零股一并卖出, 不向下取整 + assert r["items"][0]["action"] == pl.A_EXIT + + pos2 = dict(ts_code="600000.SH", price=10.0, total_qty=20000, base_qty=10000) + r2 = pl.plan_reduce_stock(ts_code="600000.SH", target_pct=0.05, position=pos2, + scale=2_000_000) + assert r2["items"][0]["qty"] == 10000 and r2["items"][0]["action"] == pl.A_TRIM, r2 + + r3 = pl.plan_reduce_stock(ts_code="600000.SH", target_pct=0.0, position=pos2, + scale=2_000_000) + assert r3["items"][0]["action"] == pl.A_EXIT and r3["items"][0]["qty"] == 20000, r3 + + r4 = pl.plan_reduce_stock(ts_code="600000.SH", target_pct=0.20, position=pos2, + scale=2_000_000) + assert r4["items"] == [] and r4["ok"], r4 # 未超目标不减持 + + +@case("组合命令·一键清仓(市值大的先卖)/行业清仓/行业限额/暂停买入撤单") +def _(): + ps = _positions_demo() + for p, s in zip(ps, ["银行", "银行", "白酒", "新能源"]): + p["sector"] = s + r = pl.plan_liquidate_all(positions=ps, pending_buys=[ + {"instruction_id": "I1", "ts_code": "600036.SH", "qty": 100, "amount": 1000}]) + assert r["items"][0]["action"] == pl.A_HALT + exits = [i for i in r["items"] if i["action"] == pl.A_EXIT] + # 市值降序: A 14万 → D/C 各 10万 (同值按代码升序) → B 8万 + assert [i["ts_code"] for i in exits] == ["600000.SH", "300750.SZ", "600519.SH", "000001.SZ"] + assert abs(r["planned_amount"] - 420_000) < 1e-6, r + + r2 = pl.plan_sector_exit(sector="银行", positions=ps) + assert {i["ts_code"] for i in r2["items"]} == {"600000.SH", "000001.SZ"}, r2 + assert abs(r2["planned_amount"] - 220_000) < 1e-6 + + # 银行占比 220/420 = 52.4% > 40% → 需减 52,000 元 + r3 = pl.plan_sector_cap(sector="银行", cap=0.40, positions=ps) + assert r3["items"] and all(i["action"] == pl.A_TRIM for i in r3["items"]), r3 + assert abs(r3["target_amount"] - 52_000) < 1e-6, r3 + r4 = pl.plan_sector_cap(sector="白酒", cap=0.40, positions=ps) + assert r4["items"] == [] and "未超上限" in r4["notes"][0], r4 + + r5 = pl.plan_halt_buy(pending_buys=[{"instruction_id": "I1", "ts_code": "600000.SH", + "qty": 100, "amount": 1000}]) + assert len(r5["items"]) == 1 and r5["items"][0]["action"] == pl.A_HALT + assert pl.plan_halt_buy(pending_buys=[])["items"] == [] + + +# ================================================================ recon +@case("回放认领·FIFO 跨指令拆分 (一笔 4000 股成交吃两条在途指令)") +def _(): + instrs = [{"instruction_id": "INS_A", "ts_code": "600000.SH", "side": "buy", "qty": 3000, + "exec_qty": 0, "action": "OPEN", "dispatched_at": "2026-07-27 09:35:00"}, + {"instruction_id": "INS_B", "ts_code": "600000.SH", "side": "buy", "qty": 2000, + "exec_qty": 0, "action": "ADD", "dispatched_at": "2026-07-27 10:05:00"}] + fills = [{"order_id": 1, "ts_code": "600000.SH", "side": "buy", "qty": 4000, + "price": 10.0, "done_time": "2026-07-27 10:30:00"}] + r = rc.map_fills_to_book(fills, instrs) + assert [a["instruction_id"] for a in r["actions"]] == ["INS_A", "INS_B"], r["actions"] + assert [a["qty"] for a in r["actions"]] == [3000, 1000], r["actions"] + assert [a["lot_type"] for a in r["actions"]] == ["BASE", "ADD"], r["actions"] + assert r["alerts"] == [] + + +@case("回放认领·下发晚于成交的指令不认领 (时间守卫)") +def _(): + instrs = [{"instruction_id": "INS_LATE", "ts_code": "600000.SH", "side": "buy", "qty": 1000, + "exec_qty": 0, "action": "OPEN", "dispatched_at": "2026-07-27 14:00:00"}] + fills = [{"order_id": 7, "ts_code": "600000.SH", "side": "buy", "qty": 1000, "price": 10.0, + "done_time": "2026-07-27 09:40:00"}] + r = rc.map_fills_to_book(fills, instrs) + assert r["actions"][0]["instruction_id"] is None, r["actions"] + assert r["alerts"] and r["alerts"][0]["code"] == rc.ALERT_EXTERNAL + + +@case("回放·外部成交并入 BASE 并告警 (设计生命线条款)") +def _(): + fills = [{"order_id": 2, "ts_code": "000001.SZ", "side": "buy", "qty": 500, "price": 8.0, + "done_time": "2026-07-27 09:40:00"}] + r = rc.map_fills_to_book(fills, []) + a = r["actions"][0] + assert a["instruction_id"] is None and a["lot_type"] == "BASE" and a["qty"] == 500 + assert rc.ALERT_EXTERNAL in a["alerts"] and len(r["alerts"]) == 1 + + +@case("回放·成交量超指令数量 → 超出部分标 OVER_FILL") +def _(): + instrs = [{"instruction_id": "INS_A", "ts_code": "600000.SH", "side": "buy", "qty": 1000, + "exec_qty": 0, "action": "OPEN", "dispatched_at": "2026-07-27 09:30:00"}] + fills = [{"order_id": 3, "ts_code": "600000.SH", "side": "buy", "qty": 1500, "price": 10.0, + "done_time": "2026-07-27 09:50:00"}] + r = rc.map_fills_to_book(fills, instrs) + assert len(r["actions"]) == 2 and r["actions"][1]["qty"] == 500 + assert rc.ALERT_OVERFILL in r["actions"][1]["alerts"], r["actions"] + + +@case("回放·幂等 (已入账 order_id 跳过) 与游标推进") +def _(): + fills = [{"order_id": 10, "ts_code": "600000.SH", "side": "buy", "qty": 100, "price": 10.0}, + {"order_id": 11, "ts_code": "600000.SH", "side": "buy", "qty": 100, "price": 10.0}] + r = rc.map_fills_to_book(fills, [], known_order_ids={10}) + assert r["skipped"] == 1 and len(r["actions"]) == 1, r + assert rc.next_cursor(fills, cursor=5) == 11 + assert rc.next_cursor([], cursor=5) == 5 + assert rc.next_cursor(fills, cursor=99) == 99 + + +@case("回放·卖出成交按核销次序分配, 批次不足时告警留差额") +def _(): + lots = [{"lot_id": "L1", "lot_type": "BASE", "qty": 1000, "open_date": 20260701}, + {"lot_id": "L2", "lot_type": "ADD", "qty": 500, "open_date": 20260710}] + ok = rc.apply_sell_to_lots(lots, 1200) + assert [x["lot_id"] for x in ok["alloc"]] == ["L2", "L1"], ok + assert [x["qty"] for x in ok["alloc"]] == [500, 700] and ok["short"] == 0 + short = rc.apply_sell_to_lots(lots, 2000) + assert short["short"] == 500 and short["alerts"][0]["code"] == rc.ALERT_SELL_NO_LOT + + +@case("回放·成交数量与价格口径回退 (filled_* 缺失时用 order_*)") +def _(): + assert rc.fill_qty_price({"order_quantity": 700, "order_price": 9.87}) == (700, 9.87) + assert rc.fill_qty_price({"filled_qty": 300, "order_quantity": 700, + "filled_price": 9.9, "order_price": 9.87}) == (300, 9.9) + assert rc.fill_qty_price({}) == (0, 0.0) + + +@case("对账·三类差异识别 (账本缺/账本多/数量不符)") +def _(): + book = [{"ts_code": "600000.SH", "total_qty": 6000}, + {"ts_code": "000001.SZ", "total_qty": 1000}, + {"ts_code": "600519.SH", "total_qty": 300}] + ds = [{"ts_code": "600000.SH", "qty": 6000}, + {"ts_code": "000001.SZ", "qty": 0}, + {"ts_code": "300750.SZ", "qty": 800}, + {"ts_code": "600519.SH", "qty": 500}] + d = {x["ts_code"]: x for x in rc.diff_positions(book, ds)} + assert "600000.SH" not in d + assert d["000001.SZ"]["kind"] == "EXTRA_IN_BOOK" and d["000001.SZ"]["delta"] == -1000 + assert d["300750.SZ"]["kind"] == "MISSING_IN_BOOK" and d["300750.SZ"]["delta"] == 800 + assert d["600519.SH"]["kind"] == "QTY_MISMATCH" and d["600519.SH"]["delta"] == 200 + + +@case("对账·以下游为准生成修正动作 (补 RECON 批 / 冲销批次)") +def _(): + diffs = rc.diff_positions([{"ts_code": "600000.SH", "total_qty": 1000}], + [{"ts_code": "600000.SH", "qty": 1500}]) + fx = rc.build_recon_fixes(diffs, price_map={"600000.SH": 10.0}) + assert fx[0]["op"] == "ADD_RECON_LOT" and fx[0]["qty"] == 500 and not fx[0]["need_price"] + diffs2 = rc.diff_positions([{"ts_code": "600000.SH", "total_qty": 1000}], + [{"ts_code": "600000.SH", "qty": 600}]) + lots = {"600000.SH": [{"lot_id": "L1", "lot_type": "BASE", "qty": 1000, + "open_date": 20260701}]} + fx2 = rc.build_recon_fixes(diffs2, price_map={}, lots_map=lots) + assert fx2[0]["op"] == "REDUCE_LOTS" and fx2[0]["qty"] == 400 + assert fx2[0]["alloc"] == [{"lot_id": "L1", "qty": 400}], fx2 + + +@case("对账·连续不一致升级 (1-2日 WARN, 满 3 日 ERROR)") +def _(): + assert rc.recon_severity(0) == rc.SEV_OK + assert rc.recon_severity(1) == rc.SEV_WARN + assert rc.recon_severity(3) == rc.SEV_ERROR + s = rc.summarize_recon([{"kind": "QTY_MISMATCH"}, {"kind": "QTY_MISMATCH"}], 3) + assert s["diff_count"] == 2 and s["by_kind"]["QTY_MISMATCH"] == 2 + assert s["severity"] == rc.SEV_ERROR + assert rc.summarize_recon([], 3)["severity"] == rc.SEV_OK + + +@case("除权检测·10送10 判为除权; 比例不吻合判 MISMATCH 待人工") +def _(): + r = rc.detect_ex_right(1000, 2000, 20.0, 10.0) + assert r["kind"] == "EX_RIGHT" and abs(r["ratio"] - 2.0) < 1e-6, r + assert rc.detect_ex_right(1000, 1000, 20.0, 20.0) is None + bad = rc.detect_ex_right(1000, 2000, 20.0, 18.0) + assert bad["kind"] == "MISMATCH" and "待人工" in bad["reason"], bad + nopx = rc.detect_ex_right(1000, 2000, 0, 10.0) + assert nopx["kind"] == "MISMATCH", nopx + # 10 送 3 (容差内) + r2 = rc.detect_ex_right(1000, 1300, 13.0, 10.0) + assert r2["kind"] == "EX_RIGHT" and abs(r2["ratio"] - 1.3) < 1e-6, r2 + + +@case("除权调整·批次数量与成本按比例调整, 总成本不变") +def _(): + lots = [{"lot_id": "L1", "lot_type": "BASE", "qty": 1000, "open_price": 20.0}, + {"lot_id": "L2", "lot_type": "ADD", "qty": 500, "open_price": 22.0}] + out = rc.apply_ex_right(lots, 2.0) + assert [l["qty"] for l in out] == [2000, 1000] + assert [l["open_price"] for l in out] == [10.0, 11.0] + before = sum(l["qty"] * l["open_price"] for l in lots) + after = sum(l["qty"] * l["open_price"] for l in out) + assert abs(before - after) < 1e-6, (before, after) + + +@case("T+1 可用量·日初重置 / 当日买入不增可卖 / 卖出扣减") +def _(): + assert rc.daily_avail_reset(6000) == 6000 + assert rc.avail_after_fill(6000, "buy", 2000) == 6000 # 当日买入 T+1 才可卖 + assert rc.avail_after_fill(6000, "sell", 2000) == 4000 + assert rc.avail_after_fill(1000, "sell", 5000) == 0 + assert rc.sellable_today(4000, 10000) == 4000 + assert rc.sellable_today(0, 100) == 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() diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py new file mode 100644 index 0000000..51896b6 --- /dev/null +++ b/scripts/test_wiring.py @@ -0,0 +1,725 @@ +# -*- coding: utf-8 -*- +""" +装配自检 (零外部依赖, 不连库) +============================== +运行: python scripts/test_wiring.py + +用内存桩替换 repo/行情, 验证「服务层 → 核心逻辑 → 落表」整条链路的接线是否正确: +导入链、API 路由、调度表与守卫、单表访问守卫、交易日历、参数中心回退, +以及命令下达→方案落表→撤销、成交回放→批次入账→成本重算 两条主干流程。 +""" +import os +import sys +import traceback +from datetime import date, datetime + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +RESULTS = [] + + +def case(name): + def deco(fn): + RESULTS.append((name, fn)) + return fn + return deco + + +# ================================================================ 内存桩 +class FakeRepo: + """pms_repo 的内存替身 (只实现服务层用到的函数, 语义与 SQL 版一致)。""" + + def __init__(self): + self.params, self.commands, self.plans = {}, {}, [] + self.positions, self.lots, self.instructions = {}, [], {} + self.proposals, self.ledger, self.reports, self.industry = {}, [], {}, {} + self._lot_id = 0 + + # --- runtime param --- + def all_params(self): + return {k: {"param_value": v, "updated_by": "test", "updated_at": ""} + for k, v in self.params.items()} + + def set_param(self, key, value, updated_by="user"): + self.params[key] = str(value) + return 1 + + def get_param(self, key): + return self.params.get(key) + + # --- command --- + def next_command_seq(self, ymd): + return len(self.commands) + 1 + + def insert_command(self, **kw): + self.commands[kw["command_id"]] = { + "id": len(self.commands) + 1, "command_id": kw["command_id"], + "cmd_class": kw["cmd_class"], "cmd_type": kw["cmd_type"], "ts_code": kw["ts_code"], + "params": kw["params"], "status": kw["status"], "progress": kw.get("progress") or {}, + "issued_by": kw.get("issued_by"), "issued_at": datetime.now(), "done_at": None, + "note": kw.get("note")} + return 1 + + def get_command(self, cid): + return self.commands.get(cid) + + def list_commands(self, *, statuses=None, cmd_class=None, limit=200): + out = [c for c in self.commands.values() + if (not statuses or c["status"] in statuses) + and (not cmd_class or c["cmd_class"] == cmd_class)] + return sorted(out, key=lambda c: -c["id"])[:limit] + + def update_command(self, cid, *, status=None, progress=None, done_at=None, note=None): + c = self.commands.get(cid) + if not c: + return 0 + if status is not None: + c["status"] = status + if progress is not None: + c["progress"] = progress + if done_at is not None: + c["done_at"] = done_at + if note is not None: + c["note"] = note + return 1 + + def supersede_param_commands(self, cmd_type, ts_code=None, keep_command_id=None): + n = 0 + for c in self.commands.values(): + if (c["cmd_class"] == "param" and c["status"] == "EFFECTIVE" + and c["cmd_type"] == cmd_type and c["ts_code"] == ts_code + and c["command_id"] != keep_command_id): + c["status"] = "SUPERSEDED" + n += 1 + return n + + def latest_effective(self, cmd_type, ts_code=None): + hits = [c for c in self.commands.values() + if c["cmd_class"] == "param" and c["status"] == "EFFECTIVE" + and c["cmd_type"] == cmd_type and (not ts_code or c["ts_code"] == ts_code)] + return sorted(hits, key=lambda c: c["id"])[-1] if hits else None + + def list_effective_stock_params(self, cmd_types=None): + return [c for c in self.commands.values() + if c["cmd_class"] == "param" and c["status"] == "EFFECTIVE" and c["ts_code"] + and (not cmd_types or c["cmd_type"] in cmd_types)] + + # --- plan --- + def insert_plans(self, rows): + for r in rows: + d = dict(r) + d.setdefault("filled_qty", 0) + self.plans.append(d) + return len(rows) + + def list_plans(self, *, command_id=None, statuses=None, ts_code=None, limit=500): + out = [p for p in self.plans + if (not command_id or p["command_id"] == command_id) + and (not statuses or p["status"] in statuses) + and (not ts_code or p["ts_code"] == ts_code)] + return sorted(out, key=lambda p: p.get("priority", 100))[:limit] + + def update_plan(self, plan_id, *, status=None, filled_qty=None): + for p in self.plans: + if p["plan_id"] == plan_id: + if status: + p["status"] = status + if filled_qty is not None: + p["filled_qty"] = filled_qty + return 1 + return 0 + + def cancel_plans_of_command(self, cid): + n = 0 + for p in self.plans: + if p["command_id"] == cid and p["status"] in ("PENDING", "EXECUTING"): + p["status"] = "CANCELLED" + n += 1 + return n + + def set_plans_deadline(self, cid, dl): + for p in self.plans: + if p["command_id"] == cid: + p["deadline"] = dl + return 1 + + # --- position / lot --- + def list_positions(self, *, only_open=False): + out = list(self.positions.values()) + return [p for p in out if int(p.get("total_qty") or 0) > 0] if only_open else out + + def get_position(self, code): + return self.positions.get(code) + + def ensure_position(self, code): + self.positions.setdefault(code, {"ts_code": code, "status": "PLANNED", "total_qty": 0, + "avail_qty": 0, "frozen_reason": "NONE"}) + return 1 + + def update_position(self, code, **fields): + self.ensure_position(code) + self.positions[code].update(fields) + return 1 + + def bump_position_qty(self, code, *, total_delta=0, avail_delta=0): + self.ensure_position(code) + p = self.positions[code] + p["total_qty"] = max(0, int(p.get("total_qty") or 0) + total_delta) + p["avail_qty"] = max(0, int(p.get("avail_qty") or 0) + avail_delta) + return 1 + + def reset_avail_all(self): + for p in self.positions.values(): + p["avail_qty"] = p.get("total_qty", 0) + p["t0_count_today"] = 0 + return len(self.positions) + + def list_lots(self, ts_code=None, *, status="OPEN", limit=1000): + out = [l for l in self.lots + if (not ts_code or l["ts_code"] == ts_code) + and (not status or l["status"] == status)] + return sorted(out, key=lambda l: (str(l["open_date"]), l["id"]))[:limit] + + def insert_lot(self, *, ts_code, lot_type, qty, open_price, open_date, instruction_id=None, + note=None): + self._lot_id += 1 + self.lots.append({"id": self._lot_id, "ts_code": ts_code, "lot_type": lot_type, + "qty": int(qty), "open_price": float(open_price), + "open_date": open_date, "closed_qty": 0, "close_avg_price": None, + "realized_pnl": 0.0, "status": "OPEN", + "instruction_id": instruction_id, "note": note}) + return 1 + + def close_lot_qty(self, lot_id, *, qty, close_price, realized_pnl): + for l in self.lots: + if l["id"] == lot_id: + old_closed = int(l["closed_qty"]) + l["close_avg_price"] = ((float(l["close_avg_price"] or 0) * old_closed + + close_price * qty) / (old_closed + qty)) + l["qty"] = max(0, int(l["qty"]) - qty) + l["closed_qty"] = old_closed + qty + l["realized_pnl"] += realized_pnl + l["status"] = "CLOSED" if l["qty"] <= 0 else "OPEN" + return 1 + return 0 + + def update_lot(self, lot_id, **fields): + for l in self.lots: + if l["id"] == lot_id: + l.update(fields) + return 1 + return 0 + + # --- instruction / proposal / ledger / report / industry --- + def insert_instruction(self, **kw): + kw.setdefault("exec_qty", 0) + kw["created_at"] = kw["updated_at"] = datetime.now() + self.instructions[kw["instruction_id"]] = kw + return 1 + + def list_instructions(self, *, statuses=None, side=None, ts_code=None, limit=300): + out = [i for i in self.instructions.values() + if (not statuses or i["status"] in statuses) + and (not side or i.get("side") == side) + and (not ts_code or i["ts_code"] == ts_code)] + return out[:limit] + + def get_instruction(self, iid): + return self.instructions.get(iid) + + def update_instruction(self, iid, **kw): + i = self.instructions.get(iid) + if not i: + return 0 + for k, v in kw.items(): + if v is not None: + i[k] = v + i["updated_at"] = datetime.now() + return 1 + + def add_instruction_exec(self, iid, qty): + i = self.instructions.get(iid) + if i: + i["exec_qty"] = int(i.get("exec_qty") or 0) + int(qty) + return 1 + + def insert_proposal(self, **kw): + kw["status"] = kw.get("status", "WAIT_USER") + kw["hard_numbers"] = kw.get("hard_numbers") or {} + self.proposals[kw["proposal_id"]] = kw + return 1 + + def list_proposals(self, *, statuses=("WAIT_USER",), limit=200): + return [p for p in self.proposals.values() if p["status"] in statuses][:limit] + + def get_proposal(self, pid): + return self.proposals.get(pid) + + def decide_proposal(self, pid, status): + p = self.proposals.get(pid) + if p and p["status"] == "WAIT_USER": + p["status"] = status + return 1 + return 0 + + def expire_proposals(self, now=None): + return 0 + + def insert_ledger(self, **kw): + self.ledger.append(kw) + return 1 + + def list_ledger(self, *, ts_code=None, limit=200): + return self.ledger[-limit:] + + def upsert_report(self, ymd, report): + self.reports[int(ymd)] = report + return 1 + + def get_report(self, ymd): + r = self.reports.get(int(ymd)) + return {"ymd": ymd, "report": r} if r else None + + def latest_report(self): + if not self.reports: + return None + y = max(self.reports) + return {"ymd": y, "report": self.reports[y]} + + def get_industry(self, code): + return self.industry.get(code) + + def list_industry(self, limit=5000): + return [{"ts_code": k, "industry": v} for k, v in self.industry.items()] + + def upsert_industry(self, rows): + for r in rows: + self.industry[r["ts_code"]] = r["industry"] + return len(rows) + + +def install_fakes(prices=None, positions=None, params=None): + """把内存桩装到各模块上, 返回 FakeRepo 实例。""" + from app.repo import downstream_repo, pms_repo + from app.services import industry, market, param_store, portfolio + + fake = FakeRepo() + fake.params.update(params or {}) + for p in (positions or []): + base = {"ts_code": p["ts_code"], "status": "HOLDING", "frozen_reason": "NONE", + "total_qty": 0, "avail_qty": 0, "base_qty": 0, "fill_qty": 0, "add_qty": 0, + "dca_qty": 0, "t0_qty": 0, "avg_cost": None, "cushion_peak": 0, + "realized_t_profit": 0, "t0_enabled": 0} + base.update(p) + fake.positions[p["ts_code"]] = base + + for mod in (pms_repo,): + for name in dir(FakeRepo): + if name.startswith("_"): + continue + setattr(mod, name, getattr(fake, name)) + market.get_price = lambda c: (prices or {}).get(c) + market.get_prices = lambda cs_: {c: (prices or {}).get(c) for c in (cs_ or [])} + market.get_refs = lambda c, **kw: {"support": None, "pressure": None, "stop": None, + "source": "none"} + industry.get_many = lambda cs_: {c: None for c in (cs_ or [])} + industry.get = lambda c: None + industry.ready = lambda: False + industry.status = lambda: {"source": "", "ready": False, "count": 0, "hint": "test"} + # 下游只读表默认桩 (单测不触网; 个别用例内再按需覆盖) + downstream_repo.fetch_buy_plans = lambda **kw: [] + downstream_repo.fetch_filled_orders = lambda **kw: [] + downstream_repo.fetch_positions = lambda: {"rows": [], "columns": {"qty": None}, + "raw_count": 0} + downstream_repo.fetch_refs = lambda c: None + param_store._cache.update({"at": 0.0, "data": {}, "error": None}) + portfolio.save_neg_streak({}) + return fake + + +# ================================================================ 用例 +@case("装配·全部模块可导入 (无循环依赖)") +def _(): + import importlib + for m in ["app.core.sizer", "app.core.cushion", "app.core.command_spec", "app.core.planner", + "app.core.recon", "app.core.tradedays", "app.db.session", "app.repo.pms_repo", + "app.repo.downstream_repo", "app.services.param_store", "app.services.industry", + "app.services.market", "app.services.portfolio", "app.services.command_service", + "app.services.ledger_service", "app.web.main", "app.scheduler"]: + importlib.import_module(m) + + +@case("装配·API 路由齐全 (四块页面 + 运维)") +def _(): + from app.web.main import app + paths = {r.path for r in app.routes} + need = ["/", "/health", "/api/overview", "/api/params", "/api/commands", + "/api/commands/catalog", "/api/commands/active", "/api/commands/{command_id}", + "/api/commands/{command_id}/cancel", "/api/plans", "/api/positions", + "/api/positions/{ts_code}/lots", "/api/instructions", "/api/ledger", "/api/report", + "/api/proposals", "/api/proposals/{proposal_id}/decide", "/api/ops/replay", + "/api/ops/reconcile", "/api/ops/daily-settle", "/api/ops/report", + "/api/ops/downstream-schema", "/api/industry"] + missing = [p for p in need if p not in paths] + assert not missing, missing + import os as _os + from app.web import main as web + assert _os.path.exists(_os.path.join(web.STATIC_DIR, "index.html")), "单页文件缺失" + + +@case("装配·调度表覆盖设计 §10 全部调度位") +def _(): + from app import scheduler as sch + names = set(sch.celery_app.conf.beat_schedule) + assert names == {"premarket", "command_poll", "replay_fills", "intraday_exec", + "signal_digest", "t0_close", "daily_settle", "daily_report"}, names + tasks = set(sch.celery_app.tasks) + for t in ("pms.premarket", "pms.command_poll", "pms.replay_fills", "pms.daily_settle", + "pms.daily_report", "pms.t0_close", "pms.intraday_exec", "pms.signal_digest"): + assert t in tasks, t + assert sch.in_session(datetime(2026, 7, 27, 10, 0)) is True + assert sch.in_session(datetime(2026, 7, 27, 12, 0)) is False + assert sch.in_session(datetime(2026, 7, 27, 14, 59)) is True + + +@case("装配·调度守卫: 非交易日跳过 / 休假模式跳过 / 异常不外抛") +def _(): + from app import scheduler as sch + from app.core import tradedays as tdm + orig = tdm.is_trade_day + try: + tdm.is_trade_day = lambda d=None: False + + @sch.guard(trade_day=True) + def t1(): + raise AssertionError("非交易日不应执行") + assert t1() == {"skipped": "not_trade_day"} + + tdm.is_trade_day = lambda d=None: True + install_fakes(params={"PMS_GLOBAL_EXEC_HALT": "true"}) + + @sch.guard(trade_day=True) + def t2(): + raise AssertionError("休假模式不应执行") + assert t2() == {"skipped": "exec_halt"} + + install_fakes(params={}) + + @sch.guard(trade_day=True) + def t3(): + raise ValueError("boom") + r = t3() + assert "error" in r and "boom" in r["error"], r + finally: + tdm.is_trade_day = orig + + +@case("装配·严格单表访问守卫 (JOIN 与多表一律拒绝)") +def _(): + from app.db.session import MultiTableSQL, assert_single_table + assert_single_table("SELECT * FROM pms_position WHERE ts_code = :c") + assert_single_table("UPDATE pms_lot SET qty = 0 WHERE id = :id") + for bad in ["SELECT a.* FROM pms_position a JOIN pms_lot b ON a.ts_code = b.ts_code", + "SELECT * FROM pms_position, pms_lot", + "SELECT * FROM pms_plan WHERE ts_code IN (SELECT ts_code FROM pms_position)"]: + try: + assert_single_table(bad) + assert False, f"未拦截: {bad}" + except MultiTableSQL: + pass + + +@case("交易日历·执行窗口与剩余交易日 (跨周末)") +def _(): + from app.core import tradedays as tdm + fri, mon = date(2026, 7, 24), date(2026, 7, 27) # 2026-07-24 周五, 07-27 周一 + assert tdm.is_trade_day(fri) and tdm.is_trade_day(mon) + assert tdm.is_trade_day(date(2026, 7, 25)) is False # 周六 + assert tdm.next_trade_day(fri, 1) == mon + assert tdm.window_deadline(mon, 1) == mon + assert tdm.window_deadline(mon, 3) == date(2026, 7, 29) + assert tdm.window_deadline(fri, 2) == mon # 周五起 2 个交易日 = 周一 + assert tdm.trade_days_left(date(2026, 7, 29), mon) == 3 + assert tdm.trade_days_left(date(2026, 7, 20), mon) == 0 + assert tdm.ymd(mon) == 20260727 + + +@case("参数中心·表值优先/文件回退/非法值拒绝") +def _(): + from app.services import param_store + from config.settings import settings + fake = install_fakes(params={"PMS_TOTAL_SCALE": "3000000", "PMS_AUTONOMY": "full"}) + assert param_store.get("PMS_TOTAL_SCALE") == 3_000_000.0 + assert param_store.get("PMS_AUTONOMY") == "full" + assert param_store.get("PMS_STOCK_CAP") == settings.PMS_STOCK_CAP # 未改写 → 文件初值 + assert param_store.get_bool("PMS_GLOBAL_BUY_HALT") is False # 运行态默认 + assert param_store.set_param("PROXY_DB_URL", "x")["ok"] is False # 基础设施不可改 + assert param_store.set_param("PMS_PORTFOLIO_CAP", 1.5)["ok"] is False # 越界 + assert param_store.set_param("PMS_AUTONOMY", "auto")["ok"] is False # 枚举非法 + assert param_store.set_param("PMS_PORTFOLIO_CAP", 0.5)["ok"] is True + assert fake.params["PMS_PORTFOLIO_CAP"] == "0.5" + assert param_store.get("PMS_PORTFOLIO_CAP") == 0.5 + sp = param_store.sizing_params() + assert sp["scale"] == 3_000_000.0 and sp["portfolio_cap"] == 0.5 + + +@case("命令服务·参数命令立即生效并覆盖旧记录") +def _(): + from app.services import command_service as csvc, param_store + fake = install_fakes() + r1 = csvc.issue("SET_SCALE", {"scale": 2_000_000}) + assert r1["ok"] and r1["status"] == "EFFECTIVE", r1 + assert param_store.get("PMS_TOTAL_SCALE") == 2_000_000.0 + r2 = csvc.issue("SET_SCALE", {"scale": 2_500_000}) + assert r2["ok"] and param_store.get("PMS_TOTAL_SCALE") == 2_500_000.0 + assert fake.commands[r1["command_id"]]["status"] == "SUPERSEDED" + assert fake.commands[r2["command_id"]]["status"] == "EFFECTIVE" + bad = csvc.issue("SET_SCALE", {"scale": -1}) + assert not bad["ok"] and any("OUT_OF_RANGE" in e for e in bad["errors"]), bad + + +@case("命令服务·个股参数命令投影到账本 (冻结/做T/止损价)") +def _(): + from app.services import command_service as csvc + fake = install_fakes(positions=[{"ts_code": "600000.SH", "total_qty": 1000, + "avg_cost": 10.0}]) + assert csvc.issue("FREEZE_STOCK", {"ts_code": "600000.SH"})["ok"] + assert fake.positions["600000.SH"]["frozen_reason"] == "COMMAND_HALT" + assert csvc.issue("T0_ENABLE", {"ts_code": "600000.SH", "t_ratio": "30%"})["ok"] + assert fake.positions["600000.SH"]["t0_enabled"] == 1 + assert abs(float(fake.positions["600000.SH"]["t0_ratio"]) - 0.30) < 1e-9 + assert csvc.issue("SET_STOP_PRICE", {"ts_code": "600000.SH", "price": 9.2})["ok"] + assert fake.positions["600000.SH"]["stop_ref"] == 9.2 + assert fake.positions["600000.SH"]["ref_source"] == "user" + assert csvc.issue("UNFREEZE_STOCK", {"ts_code": "600000.SH"})["ok"] + assert fake.positions["600000.SH"]["frozen_reason"] == "NONE" + sp = csvc.effective_stock_params()["600000.SH"] + assert sp["frozen"] is False and sp["t0"] is True and sp["stop_price"] == 9.2, sp + # 做T比例超硬上限 1/3 被拒 + assert not csvc.issue("T0_ENABLE", {"ts_code": "600000.SH", "t_ratio": "40%"})["ok"] + + +@case("命令服务·降仓命令端到端 (方案落表 + 进度 + 撤销)") +def _(): + from app.services import command_service as csvc + fake = install_fakes( + prices={"600000.SH": 10.0, "000001.SZ": 8.0}, + params={"PMS_TOTAL_SCALE": "2000000", "PMS_WEAK_NEG_DAYS": "5"}, + positions=[{"ts_code": "600000.SH", "total_qty": 14000, "base_qty": 7000, + "fill_qty": 3500, "add_qty": 3500, "avg_cost": 8.93}, + {"ts_code": "000001.SZ", "total_qty": 10000, "base_qty": 10000, + "avg_cost": 8.8}]) + from app.services import portfolio + portfolio.save_neg_streak({"000001.SZ": 6}) + + r = csvc.issue("REDUCE_EXPOSURE", {"pct": "5%", "window_tdays": 3}) + assert r["ok"] and r["status"] == "EXECUTING", r + cid = r["command_id"] + plans = fake.list_plans(command_id=cid) + assert plans, "方案未落表" + assert {p["action"] for p in plans} <= {"EXIT", "TRIM", "HALT"}, plans + exit_p = [p for p in plans if p["action"] == "EXIT"] + assert exit_p and exit_p[0]["ts_code"] == "000001.SZ", plans # 弱票优先清 + assert all(p["deadline"] for p in plans) + prog = fake.commands[cid]["progress"] + assert prog["target_amount"] == 100_000.0 and prog["plan_count"] == len(plans), prog + + # 进度结算: 方案未成交 → 保持 EXECUTING; 全部成交 → DONE + r2 = csvc.refresh_progress(cid) + assert r2["commands"][0]["status"] in ("EXECUTING", "PARTIAL"), r2 + for p in plans: + fake.update_plan(p["plan_id"], status="DONE", filled_qty=p["qty"]) + csvc.refresh_progress(cid) + assert fake.commands[cid]["status"] == "DONE", fake.commands[cid] + + # 撤销: 已完成命令不可撤 + assert csvc.cancel(cid)["ok"] is False + r3 = csvc.issue("REDUCE_EXPOSURE", {"pct": "3%"}, force_conflict=True) + assert csvc.cancel(r3["command_id"])["ok"] is True + assert fake.commands[r3["command_id"]]["status"] == "CANCELLED" + assert all(p["status"] == "CANCELLED" + for p in fake.list_plans(command_id=r3["command_id"])) + + +@case("命令服务·冲突拦截与强制下达") +def _(): + from app.services import command_service as csvc + install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"}, + positions=[{"ts_code": "600000.SH", "total_qty": 14000, "base_qty": 7000, + "avg_cost": 9.0}]) + a = csvc.issue("REDUCE_EXPOSURE", {"pct": "5%"}) + assert a["ok"], a + b = csvc.issue("INCREASE_EXPOSURE", {"pct": "5%"}) + assert not b["ok"] and b["conflicts"], b + assert any("方向相反" in c["reason"] for c in b["conflicts"]) + c = csvc.issue("INCREASE_EXPOSURE", {"pct": "5%"}, force_conflict=True) + assert c["ok"] and c["conflicts"], c # 强制下达但冲突仍留痕 + + +@case("命令服务·全局暂停买入撤在途买入指令并置开关") +def _(): + from app.services import command_service as csvc, param_store + fake = install_fakes(prices={"600000.SH": 10.0}) + fake.insert_instruction(instruction_id="INS_1", origin_type="plan", origin_id="P1", + ts_code="600000.SH", action="OPEN", side="buy", qty=1000, + limit_price=10.0, status="DISPATCHED") + fake.insert_instruction(instruction_id="INS_2", origin_type="plan", origin_id="P2", + ts_code="600000.SH", action="EXIT", side="sell", qty=500, + limit_price=10.0, status="DISPATCHED") + r = csvc.issue("HALT_BUY", {}) + assert r["ok"] and r["status"] == "DONE", r + assert fake.instructions["INS_1"]["status"] == "CANCELLED" + assert fake.instructions["INS_2"]["status"] == "DISPATCHED" # 卖出不受影响 + assert param_store.get_bool("PMS_GLOBAL_BUY_HALT") is True + assert csvc.issue("RESUME_BUY", {})["ok"] + assert param_store.get_bool("PMS_GLOBAL_BUY_HALT") is False + + +@case("账本服务·成交回放入账 → 批次/持仓/成本重算") +def _(): + from app.services import ledger_service as ls + from app.repo import downstream_repo + fake = install_fakes(prices={"600000.SH": 11.0}) + fake.insert_instruction(instruction_id="INS_A", origin_type="plan", origin_id="P1", + ts_code="600000.SH", action="OPEN", side="buy", qty=6000, + limit_price=10.0, status="DISPATCHED") + fills = [{"order_id": 101, "ts_code": "600000.SH", "side": "buy", "qty": 6000, + "price": 10.0, "done_time": "2026-07-27 09:40:00"}] + orig = downstream_repo.fetch_filled_orders + try: + downstream_repo.fetch_filled_orders = lambda **kw: fills + r = ls.replay_fills() + assert r["ok"] and r["actions"] == 1, r + assert fake.positions["600000.SH"]["total_qty"] == 6000 + assert fake.positions["600000.SH"]["avail_qty"] == 0 # T+1: 当日买入不可卖 + assert abs(float(fake.positions["600000.SH"]["avg_cost"]) - 10.0) < 1e-6 + assert fake.positions["600000.SH"]["base_qty"] == 6000 + assert abs(float(fake.positions["600000.SH"]["cushion_pct"]) - 0.10) < 1e-4 + assert fake.instructions["INS_A"]["status"] == "CONFIRMED" + assert fake.params["PMS_REPLAY_CURSOR"] == "101" + # 幂等: 游标已推进, 同一批不再重复入账 + downstream_repo.fetch_filled_orders = lambda **kw: [] + r2 = ls.replay_fills() + assert r2["fills"] == 0 and fake.positions["600000.SH"]["total_qty"] == 6000 + finally: + downstream_repo.fetch_filled_orders = orig + + +@case("账本服务·卖出回放按核销次序 + 摊薄成本下降") +def _(): + from app.services import ledger_service as ls + from app.repo import downstream_repo + fake = install_fakes(prices={"600000.SH": 12.0}) + fake.insert_lot(ts_code="600000.SH", lot_type="BASE", qty=6000, open_price=10.0, + open_date="2026-07-01") + fake.insert_lot(ts_code="600000.SH", lot_type="ADD", qty=3000, open_price=11.0, + open_date="2026-07-20") + fake.update_position("600000.SH", total_qty=9000, avail_qty=9000) + fills = [{"order_id": 201, "ts_code": "600000.SH", "side": "sell", "qty": 3000, + "price": 12.0, "done_time": "2026-07-27 10:00:00"}] + orig = downstream_repo.fetch_filled_orders + try: + downstream_repo.fetch_filled_orders = lambda **kw: fills + r = ls.replay_fills() + assert r["ok"], r + add_lot = [l for l in fake.lots if l["lot_type"] == "ADD"][0] + assert add_lot["qty"] == 0 and add_lot["status"] == "CLOSED" # ADD 先核销 + base_lot = [l for l in fake.lots if l["lot_type"] == "BASE"][0] + assert base_lot["qty"] == 6000 # 底仓保留 + p = fake.positions["600000.SH"] + assert p["total_qty"] == 6000 and p["avail_qty"] == 6000 + # 摊薄成本 = (60000+33000-36000)/6000 = 9.5 + assert abs(float(p["avg_cost"]) - 9.5) < 1e-6, p["avg_cost"] + finally: + downstream_repo.fetch_filled_orders = orig + + +@case("账本服务·对账以下游为准 + 连续不一致升级") +def _(): + from app.services import ledger_service as ls + from app.repo import downstream_repo + fake = install_fakes(prices={"600000.SH": 10.0}) + fake.insert_lot(ts_code="600000.SH", lot_type="BASE", qty=1000, open_price=10.0, + open_date="2026-07-01") + fake.update_position("600000.SH", total_qty=1000, avail_qty=1000) + orig = downstream_repo.fetch_positions + try: + downstream_repo.fetch_positions = lambda: { + "rows": [{"ts_code": "600000.SH", "qty": 1500, "avail_qty": 1500}], + "columns": {"code": "stock_code", "qty": "current_qty"}, "raw_count": 1} + r = ls.reconcile() + assert r["ok"] and len(r["diffs"]) == 1 and r["diffs"][0]["delta"] == 500, r + assert fake.positions["600000.SH"]["total_qty"] == 1500 # 以下游为准 + assert any(l["lot_type"] == "RECON" for l in fake.lots) # 修正留痕 + assert any(x.get("action") == "RECON" for x in fake.ledger) + assert r["severity"] == "WARN" + for _i in range(2): # 连续第 3 日 → ERROR + downstream_repo.fetch_positions = lambda: { + "rows": [{"ts_code": "600000.SH", "qty": 1500 + 100 * (_i + 1)}], + "columns": {"qty": "current_qty"}, "raw_count": 1} + r = ls.reconcile() + assert r["severity"] == "ERROR", r + finally: + downstream_repo.fetch_positions = orig + + +@case("账本服务·日报生成 (关注区 + 次日除权检测快照)") +def _(): + from app.services import ledger_service as ls + fake = install_fakes(prices={"600000.SH": 10.0}, params={"PMS_TOTAL_SCALE": "2000000"}, + positions=[{"ts_code": "600000.SH", "total_qty": 6000, + "avg_cost": 9.0}]) + rep = ls.build_daily_report() + assert rep["totals"]["portfolio_mv"] == 60_000.0 + assert rep["snapshot"]["600000.SH"] == {"qty": 6000, "price": 10.0} + assert any(a["type"] == "行业约束停用" for a in rep["attention"]), rep["attention"] + assert fake.reports[rep["ymd"]]["ymd"] == rep["ymd"] + + +@case("账本服务·除权检测走通 (10送10 → 批次按比例调整)") +def _(): + from app.services import ledger_service as ls + fake = install_fakes(prices={"600000.SH": 5.0}, params={"PMS_TOTAL_SCALE": "2000000"}, + positions=[{"ts_code": "600000.SH", "total_qty": 2000, + "avg_cost": 10.0}]) + fake.insert_lot(ts_code="600000.SH", lot_type="BASE", qty=2000, open_price=10.0, + open_date="2026-07-01") + fake.upsert_report(20260726, {"snapshot": {"600000.SH": {"qty": 1000, "price": 10.0}}}) + r = ls.detect_and_apply_ex_right() + assert r["ex_rights"] and abs(r["ex_rights"][0]["ratio"] - 2.0) < 1e-6, r + lot = fake.lots[0] + assert lot["qty"] == 4000 and abs(lot["open_price"] - 5.0) < 1e-6, lot + + +@case("盘前准备·T+1 可卖重置") +def _(): + from app.services import ledger_service as ls + fake = install_fakes(prices={"600000.SH": 10.0}, + positions=[{"ts_code": "600000.SH", "total_qty": 6000, "avail_qty": 0, + "avg_cost": 9.0}]) + r = ls.premarket() + assert r["avail_reset"] >= 1, r + assert fake.positions["600000.SH"]["avail_qty"] == 6000 + + +# ---------------------------------------------------------------- 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()