# -*- coding: utf-8 -*- """技术面接入 · 工作包一 (2026-09-11 方案第五节, 台账 008)。全部离线, 不连库不连网。 A 相位合成 (core/tech_rules): 四种无读数、九个相位逐格、震荡市抑制翻向信号、数翻向。 B 取数解析 (tech_service.parse_item): 代码归一 SH600000→600000.SH、字段映射、认不出返回 None。 C 分页取数 (fetch_all_pages): 多页拼接、matched 终止、状态非 OK 抛错 (注入 fetch, 不连网)。 D 映射 (build_map): 正常合成只覆盖有当天读数的相关票、读数陈旧整份按无读数 (桩掉库与网络)。 E 参数与口径: PMS_TECH_* 登记进 RUNTIME_EXTRA 与 _RANGES、交易日差口径。 F 落表 (pull): 接口状态非 OK 整轮不落表、正常落表并推进日期 (桩掉库)。 """ import os import sys import traceback from datetime import datetime sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.core import tech_rules as tr # noqa: E402 from app.services import param_store # noqa: E402 from app.services import tech_service as ts # noqa: E402 RESULTS = [] def case(name): def deco(fn): RESULTS.append((name, fn)) return fn return deco class _Patch: def __init__(self): self._saved = [] def __enter__(self): return self def __call__(self, obj, name, value): self._saved.append((obj, name, getattr(obj, name))); setattr(obj, name, value) def __exit__(self, *a): for obj, name, v in reversed(self._saved): setattr(obj, name, v) def _row(dd, *, side="多", pos=0.6, squeeze=False, bw=5.0, bbi="中性区", flip=8, quality="OK", reanchor=1, sar=10.0): """造一行 pms_tech_daily 读数 (只填 synthesize 用得到的字段)。""" return {"data_date": dd, "ts_code": "600000.SH", "sar_side": side, "boll_pos": pos, "boll_squeeze": 1 if squeeze else 0, "boll_bw_pct": bw, "bbi_state": bbi, "sar_flip_days": flip, "quality": quality, "reanchored": reanchor, "sar_value": sar} # ================================================================ A 相位合成 @case("A 无读数: 没有当天行 → 弃权, 不折成看空") def test_noread_empty(): r = tr.synthesize([]) assert r["stance"] == "无读数" and r["no_read_why"] @case("A 无读数: 未除权重锚") def test_noread_reanchor(): r = tr.synthesize([_row(20260910, reanchor=0)]) assert r["stance"] == "无读数" and "除权" in r["reason"] @case("A 无读数: 数据质量非 OK") def test_noread_quality(): r = tr.synthesize([_row(20260910, quality="STALE")]) assert r["stance"] == "无读数" and "质量" in r["reason"] @case("A 无读数: 关键读数缺失 (SAR 方向)") def test_noread_missing(): r = tr.synthesize([_row(20260910, side="")]) assert r["stance"] == "无读数" @case("A 收口等待: 今天收口一律中性") def test_squeeze(): r = tr.synthesize([_row(20260910, squeeze=True, side="多", bbi="多头区")]) assert r["phase"] == "收口等待" and r["stance"] == "中性" @case("A 趋势多: SAR 多 + 多空布林线多头区 = 看多强") def test_trend_long_strong(): r = tr.synthesize([_row(20260910, side="多", bbi="多头区", flip=8)]) assert r["phase"] == "趋势多" and r["stance"] == "看多" and r["strength"] == "强" @case("A 趋势多: SAR 多 + 中性区 = 看多弱") def test_trend_long_weak(): r = tr.synthesize([_row(20260910, side="多", bbi="中性区", flip=8)]) assert r["phase"] == "趋势多" and r["stance"] == "看多" and r["strength"] == "弱" @case("A 趋势空: SAR 空 + 空头区 = 看空强") def test_trend_short_strong(): r = tr.synthesize([_row(20260910, side="空", bbi="空头区", flip=8)]) assert r["phase"] == "趋势空" and r["stance"] == "看空" and r["strength"] == "强" @case("A 趋势空: SAR 空 + 中性区 = 看空弱") def test_trend_short_weak(): r = tr.synthesize([_row(20260910, side="空", bbi="中性区", flip=8)]) assert r["phase"] == "趋势空" and r["stance"] == "看空" and r["strength"] == "弱" @case("A 分歧: SAR 多 + 空头区 = 中性") def test_diverge_long(): r = tr.synthesize([_row(20260910, side="多", bbi="空头区", flip=8)]) assert r["phase"] == "分歧" and r["stance"] == "中性" @case("A 分歧: SAR 空 + 多头区 = 中性") def test_diverge_short(): r = tr.synthesize([_row(20260910, side="空", bbi="多头区", flip=8)]) assert r["phase"] == "分歧" and r["stance"] == "中性" @case("A 转多: SAR 刚翻多 + 多空布林线非空头 = 看多弱") def test_flip_long(): r = tr.synthesize([_row(20260910, side="多", bbi="中性区", flip=1)]) assert r["phase"] == "转多" and r["stance"] == "看多" and r["strength"] == "弱" @case("A 转空: 刚翻空 + 非空头非贴轨 = 看空弱 (未确认)") def test_flip_short_unconfirmed(): r = tr.synthesize([_row(20260910, side="空", bbi="中性区", pos=0.5, flip=1)]) assert r["phase"] == "转空" and r["confirm"] is False and r["strength"] == "弱" @case("A 转空: 刚翻空 + 空头区 = 看空强 (确认)") def test_flip_short_confirmed(): r = tr.synthesize([_row(20260910, side="空", bbi="空头区", flip=1)]) assert r["phase"] == "转空" and r["confirm"] is True and r["strength"] == "强" @case("A 开口向上: 收口后放开 + 上半部 + SAR 多 = 看多强") def test_open_up(): rows = [_row(20260909, squeeze=True, bw=5.0, side="多", bbi="中性区", flip=8), _row(20260910, squeeze=False, bw=6.2, pos=0.6, side="多", bbi="中性区", flip=8)] r = tr.synthesize(rows) assert r["phase"] == "开口向上" and r["stance"] == "看多" and r["strength"] == "强" @case("A 开口向下: 收口后放开 + 下半部 + SAR 空 = 看空强") def test_open_down(): rows = [_row(20260909, squeeze=True, bw=5.0, side="空", bbi="中性区", flip=8), _row(20260910, squeeze=False, bw=6.2, pos=0.4, side="空", bbi="中性区", flip=8)] r = tr.synthesize(rows) assert r["phase"] == "开口向下" and r["stance"] == "看空" and r["strength"] == "强" @case("A 带宽没扩够不算开口, 落回趋势") def test_open_not_enough(): rows = [_row(20260909, squeeze=True, bw=5.0, side="多", bbi="中性区", flip=8), _row(20260910, squeeze=False, bw=5.5, pos=0.6, side="多", bbi="中性区", flip=8)] r = tr.synthesize(rows) assert r["phase"] != "开口向上" and r["phase"] == "趋势多" @case("A 震荡市里刚翻空不判转空 (SAR 翻向不当信号), 按趋势走") def test_choppy_suppresses_flip(): # 20 行交替方向 → 翻向 19 次, 远超 4 → 震荡市; 今日刚翻空 (flip=1) flippy = [_row(20260800 + i, side=("多" if i % 2 == 0 else "空"), bbi="中性区", flip=1) for i in range(20)] r = tr.synthesize(flippy) assert r["choppy"] is True assert r["phase"] != "转空", "震荡市里刚翻空不该判转空" assert r["phase"] == "趋势空" and r["stance"] == "看空" @case("A count_sar_flips: 数相邻方向变化") def test_count_flips(): rows = [_row(i, side=("多" if i % 2 == 0 else "空")) for i in range(5)] assert tr.count_sar_flips(rows) == 4 assert tr.count_sar_flips([_row(0, side="多"), _row(1, side="多")]) == 0 # ================================================================ B 取数解析 @case("B parse_item: 代码归一 + 字段映射 + 认不出返回 None") def test_parse_item(): it = {"stock_code": "SH600000", "stock_name": "浦发银行", "boll": {"upper": 9.4, "mid": 9.1, "lower": 8.9, "bandwidth_pct": 5.4, "pos": 0.85, "state": "带内偏上", "squeeze": True}, "bbiboll": {"bbi": 9.2, "state": "多头区", "dist_pct": 1.0}, "sar": {"value": 9.1, "side": "多", "flip_days": 8, "dist_pct": 2.5}, "reanchored": True, "in_pool": False, "quality": "OK", "bars_used": 120, "last_bar_date": 20260910, "bars_lag": 0} r = ts.parse_item(it, 20260910, "tech_view_v1") assert r["ts_code"] == "600000.SH" and r["stock_name"] == "浦发银行" assert r["boll_squeeze"] == 1 and r["reanchored"] == 1 and r["in_pool"] == 0 assert r["sar_side"] == "多" and r["bbi_state"] == "多头区" and r["sar_flip_days"] == 8 assert r["algo_version"] == "tech_view_v1" and r["data_date"] == 20260910 assert ts.parse_item({"stock_code": ""}, 20260910) is None assert ts.parse_item("x", 20260910) is None # ================================================================ C 分页取数 @case("C fetch_all_pages: 分页拼接 + matched 终止 + 状态非 OK 抛错") def test_fetch_pages(): pages = {0: [{"stock_code": "SH600000", "sar": {"side": "多"}, "reanchored": True, "quality": "OK", "boll": {"pos": 0.5}}, {"stock_code": "SZ000001", "sar": {"side": "空"}, "reanchored": True, "quality": "OK", "boll": {"pos": 0.4}}], 2: [{"stock_code": "SH600004", "sar": {"side": "多"}, "reanchored": True, "quality": "OK", "boll": {"pos": 0.6}}]} def f(url, params, timeout): return {"status": "OK", "data_date": 20260910, "matched": 3, "algo_version": "v1", "items": pages.get(params["offset"], [])} got = ts.fetch_all_pages(base="x", path="/y", timeout=5, page_size=2, fetch=f) assert got["total"] == 3 and got["pages"] == 2, got assert {r["ts_code"] for r in got["rows"]} == {"600000.SH", "000001.SZ", "600004.SH"} def bad(url, params, timeout): return {"status": "DEGRADED", "items": []} try: ts.fetch_all_pages(base="x", path="/y", timeout=5, fetch=bad) assert False, "状态非 OK 应抛 TechFeedError" except ts.TechFeedError: pass # ================================================================ D 映射 @case("D build_map: 正常合成, 只覆盖有当天读数的相关票") def test_build_map_ok(): P = {"enabled": True, "stale_tdays": 2, "keep_days": 40, "base": "x", "path": "/y", "timeout": 20, "rules": dict(tr.DEFAULTS)} saved = {} with _Patch() as p: p(ts, "_params", lambda: P) p(ts.tech_repo, "latest_date", lambda: 20260910) p(ts, "_relevant_codes", lambda: ["600000.SH", "000001.SZ"]) p(ts.tech_repo, "history_multi", lambda codes, since=0: {"600000.SH": [_row(20260910, side="多", bbi="多头区", flip=8)]}) p(ts, "_save_map", lambda m: (saved.update(m), {"ok": True})[1]) out = ts.build_map(now=datetime(2026, 9, 11, 8, 45)) assert out["ok"] and out["data_date"] == 20260910 and out["codes"] == 2 assert out["states"] == 1, "000001 今日无读数, 不进映射" assert saved["states"]["600000.SH"]["stance"] == "看多" @case("D build_map: 读数陈旧整份按无读数, 映射置空带原因") def test_build_map_stale(): P = {"enabled": True, "stale_tdays": 2, "keep_days": 40, "base": "x", "path": "/y", "timeout": 20, "rules": dict(tr.DEFAULTS)} saved = {} with _Patch() as p: p(ts, "_params", lambda: P) p(ts.tech_repo, "latest_date", lambda: 20260901) # 距 9/11 超过 2 个交易日 p(ts, "_relevant_codes", lambda: ["600000.SH"]) p(ts, "_save_map", lambda m: (saved.update(m), {"ok": True})[1]) out = ts.build_map(now=datetime(2026, 9, 11, 8, 45)) assert not out["ok"] and out["error"] == "读数陈旧" assert saved.get("stale") is True and saved["states"] == {} # ================================================================ E 参数与口径 @case("E 参数登记: PMS_TECH_* 在 RUNTIME_EXTRA 与 _RANGES") def test_params_registered(): for k in ("PMS_TECH_ENABLED", "PMS_TECH_API_BASE", "PMS_TECH_API_PATH", "PMS_TECH_TIMEOUT", "PMS_TECH_KEEP_DAYS", "PMS_TECH_STALE_TDAYS", "PMS_TECH_SQUEEZE_LOOKBACK", "PMS_TECH_OPEN_BW_GROWTH", "PMS_TECH_CHOPPY_FLIPS", "PMS_TECH_FLIP_FRESH_DAYS", "PMS_TECH_STATE_MAP"): assert k in param_store.RUNTIME_EXTRA, k for k in ("PMS_TECH_TIMEOUT", "PMS_TECH_KEEP_DAYS", "PMS_TECH_STALE_TDAYS", "PMS_TECH_SQUEEZE_LOOKBACK", "PMS_TECH_OPEN_BW_GROWTH", "PMS_TECH_CHOPPY_FLIPS", "PMS_TECH_FLIP_FRESH_DAYS"): assert k in param_store._RANGES, k @case("E _tdays_elapsed: 昨→今 1 个交易日, 同日 0") def test_tdays(): assert ts._tdays_elapsed(20260910, 20260911) == 1 assert ts._tdays_elapsed(20260911, 20260911) == 0 # ================================================================ F 落表 @case("F pull: 接口状态非 OK 整轮不落表") def test_pull_no_ok(): P = {"enabled": True, "base": "x", "path": "/y", "timeout": 5, "keep_days": 40, "stale_tdays": 2, "rules": dict(tr.DEFAULTS)} called = {"upsert": 0} with _Patch() as p: p(ts, "_params", lambda: P) p(ts.tech_repo, "upsert_daily", lambda rows: (called.__setitem__("upsert", called["upsert"] + 1), len(rows))[1]) out = ts.pull(fetch=lambda url, params, timeout: {"status": "ERR", "items": []}) assert not out["ok"] and called["upsert"] == 0, "状态非 OK 绝不落表" @case("F pull: 正常落表并推进日期") def test_pull_ok(): P = {"enabled": True, "base": "x", "path": "/y", "timeout": 5, "keep_days": 40, "stale_tdays": 2, "rules": dict(tr.DEFAULTS)} got = {"rows": None} def good(url, params, timeout): if params["offset"] == 0: return {"status": "OK", "data_date": 20260910, "matched": 1, "algo_version": "v1", "items": [{"stock_code": "SH600000", "boll": {"squeeze": False, "pos": 0.6}, "bbiboll": {"state": "多头区"}, "sar": {"side": "多", "value": 9.1, "flip_days": 8}, "reanchored": True, "quality": "OK"}]} return {"status": "OK", "data_date": 20260910, "matched": 1, "items": []} with _Patch() as p: p(ts, "_params", lambda: P) p(ts.tech_repo, "upsert_daily", lambda rows: (got.__setitem__("rows", rows), len(rows))[1]) p(ts.tech_repo, "prune", lambda before: 0) out = ts.pull(fetch=good) assert out["ok"] and out["data_date"] == 20260910 and out["rows"] == 1 assert got["rows"][0]["ts_code"] == "600000.SH" @case("F upsert SQL: ON DUPLICATE 子句用 VALUES() 不带绑定参数（pymysql 批量陷阱, 2026-09-11 真机抓到）") def test_upsert_odku_no_bind(): import inspect from app.repo import tech_repo src = inspect.getsource(tech_repo.upsert_daily) assert "= VALUES(" in src, "ON DUPLICATE 子句要用 VALUES(列) 引用插入值" assert 'f"{c} = :{c}"' not in src, "UPDATE 子句不该再用 :col 绑定参数（pymysql executemany 多行合并会把它错位展开）" 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())