# -*- coding: utf-8 -*- """点股票看它今天全部信号 (2026-09-09 方案第三节, 台账 055)。全部离线, 不连 Redis 不连库。 1. 代码归一与匹配: 四种写法归一、跨市场同六位不误配、认不出的回错误不抛。 2. 分页回扫边界: 手工减一的流 id 边界、按今天零点截断、上限截断标记。 3. 五源解析: 决策与择时层分流、告警超龄标陈旧、风控卖出载荷在 data 里、资金异动时间取流 id。 4. 去重口径: 同源同分钟同值合并并计数、跨源永不合并。 5. 韧性: 单源失败不拖垮整体且逐源标明、按票缓存、实况分不编时间戳。 6. 只读守卫与页面静态断言。 """ import io import os import re import sys import time import traceback sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.services import param_store # noqa: E402 from app.services import upstream_signals as us # 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 _today_ms(h, m): t = time.localtime() return int(time.mktime((t.tm_year, t.tm_mon, t.tm_mday, h, m, 0, 0, 0, -1)) * 1000) class _FakeRedis: """只实现 xrevrange 的假客户端。streams: {key: [(id, fields), ...]} 按 id 升序存, 读时倒序。""" def __init__(self, streams=None, zsets=None, boom=None): self.streams = streams or {} self.zsets = zsets or {} self.boom = boom or set() self.calls = [] def xrevrange(self, key, max="+", min="-", count=None): if key in self.boom: raise RuntimeError("stream down: %s" % key) self.calls.append((key, max, count)) rows = sorted(self.streams.get(key, []), key=lambda r: _idk(r[0]), reverse=True) if max != "+": rows = [r for r in rows if _idk(r[0]) <= _idk(max)] return rows[:count] if count else rows def zrevrange(self, key, s, e, withscores=False): return self.zsets.get(key, []) def _idk(sid): a, b = str(sid).split("-") return (int(a), int(b)) @case("代码归一: 四种写法同归、跨市场同六位不误配、认不出回错误") def test_code_norm(): assert us._norm_code("SH601126") == "601126.SH" assert us._norm_code("601126.SH") == "601126.SH" assert us._norm_code("sh601126") == "601126.SH" assert us._norm_code("601126") == "601126", us._norm_code("601126") assert us._norm_code("洋垃圾") == "" and us._norm_code(None) == "" assert us._same_code("SH601126", "601126.SH") and us._same_code("601126", "601126.SZ") assert not us._same_code("601126.SH", "601126.SZ"), "带后缀且不同市场必须判不同" assert not us._same_code("601126.SH", "000001.SZ") and not us._same_code("", "601126.SH") out = us.by_code("洋垃圾") assert out["ok"] is False and out["timeline"] == [], out @case("分页回扫边界: 手工减一、翻页不重不漏、上限截断标记") def test_scan_paging(): rows = [("%d-0" % (1000 + i), {"i": i}) for i in range(12)] c = _FakeRedis({"k": rows}) got, trunc = us._scan_stream(c, "k", cap=100, page=5) assert len(got) == 12 and not trunc, (len(got), trunc) assert [f["i"] for _s, f in got] == list(range(11, -1, -1)), "必须倒序且不重不漏" assert len(c.calls) == 3 and c.calls[1][1] == "1006-18446744073709551615", c.calls got2, trunc2 = us._scan_stream(c, "k", cap=7, page=5) assert len(got2) == 7 and trunc2 is True, (len(got2), trunc2) assert us._prev_id("100-3") == "100-2" and us._prev_id("100-0") == "99-18446744073709551615" assert us._prev_id("0-0") == "-" and us._prev_id("garbage") == "-" assert us._id_ms("1757000000000-5") == 1757000000000 and us._id_ms("x") == 0 got3, trunc3 = us._scan_stream(c, "缺这条流", cap=100) assert got3 == [] and not trunc3 @case("不分键的流扫到今天零点为止, 且不算截断") def test_stop_before_today(): y = us._today_start_ms() - 3600_000 rows = [("%d-0" % y, {"a": "昨天"}), ("%d-0" % _today_ms(9, 40), {"a": "今天"})] c = _FakeRedis({"k": rows}) got, trunc = us._scan_stream(c, "k", cap=100, page=50, stop_before_ms=us._today_start_ms()) assert len(got) == 1 and got[0][1]["a"] == "今天", got assert trunc is False, "今天的已经扫全, 不能报截断" @case("五源解析: 决策与择时层分流、告警超龄标陈旧、卖出载荷在 data 里、资金异动取流 id") def test_parse_sources(): ymd = "2026-09-09" itd = [("1-0", {"ts_code": "601126.SH", "producer_id": "bionic_decider", "action": "BUY", "trigger_time": _today_ms(10, 4), "suggested_price": "43.15", "confidence": "0.95", "signal_type": "REVERSAL_BUY"}), ("2-0", {"ts_code": "601126.SH", "producer_id": "intraday_timing_v2", "action": "BUY", "trigger_time": _today_ms(10, 30), "suggested_price": "43.0", "entry_score": "78"}), ("3-0", {"ts_code": "000001.SZ", "producer_id": "bionic_decider", "action": "BUY", "trigger_time": _today_ms(10, 31)})] rows, n, _t = us._bycode_intraday(_FakeRedis({"intraday_signals:%s" % ymd: itd}), ymd, "601126.SH", 5000) assert n == 3 and len(rows) == 2, (n, len(rows)) # 流是倒序读的, 所以 10:30 的择时层那条在前 assert rows[0]["src"] == "timing" and rows[0]["entry_score"] == 78.0, rows[0] assert rows[1]["src"] == "decision" and rows[1]["time"] == "10:04" and rows[1]["confidence"] == 0.95, rows[1] assert rows[1]["cat_label"] == "建议买入" and rows[1]["price"] == 43.15 old = _today_ms(9, 35) with _Patch() as p: p(us, "_alert_max_age_sec", lambda: 60) al = [("1-0", {"ts_code": "601126.SH", "source": "fund_flow_strength", "level": "HIGH", "value": "13.41", "trigger_time": old}), ("2-0", {"ts_code": "601126.SH", "source": "未知源", "level": "MID", "value": "1", "trigger_time": int(time.time() * 1000), "metadata": '{"direction": "down"}'})] arows, an, _ = us._bycode_alerts(_FakeRedis({"intraday_alerts:%s" % ymd: al}), ymd, "SH601126", 5000) assert an == 2 and len(arows) == 2 stale = [r for r in arows if r["time"] == "09:35"][0] assert stale["stale"] is True and stale["cat"] and stale["age_min"] >= 0, stale unknown = [r for r in arows if not r["stale"]][0] assert unknown["cat"] == "other" and unknown["direction"] == "看跌", unknown sell = [("1-0", {"data": '{"ts_code": "601126.SH", "action": "SELL", "confidence": 0.8,' ' "dominant_signal": "BREAK_DOWN", "llm_reason": "跌破支撑",' ' "timestamp": %d}' % _today_ms(14, 10)})] srows, _n, _t = us._bycode_sell(_FakeRedis({"bionic:signals:llm_sell_actions": sell}), "601126.SH", 5000) assert len(srows) == 1 and srows[0]["time"] == "14:10" and srows[0]["reason"] == "跌破支撑", srows mid = _today_ms(13, 41) met = [("%d-0" % mid, {"data": '{"ts_code": "601126.SH", "direction": "up", "z_dd": 2.4, "window_net": 1200}'})] mrows, _n, _t = us._bycode_metrics(_FakeRedis({"mtf:intraday:stream:metrics": met}), "601126.SH", 5000) assert len(mrows) == 1 and mrows[0]["time"] == "13:41" and mrows[0]["z_dd"] == 2.4 assert mrows[0]["ts_from"] == "stream_id", "时间取自流 id 是写入时刻, 必须标出来" @case("去重: 同源同分钟同值合并并计数; 跨源永不合并") def test_dedup(): a = us._row("alert", "盘中告警", _today_ms(10, 4), cat="fund_flow_strength", level="HIGH", value=13.41) b = us._row("alert", "盘中告警", _today_ms(10, 4), cat="fund_flow_strength", level="HIGH", value=13.41) c = us._row("alert", "盘中告警", _today_ms(10, 4), cat="fund_flow_strength", level="HIGH", value=20.0) d = us._row("decision", "决策系统广播", _today_ms(10, 4), action="BUY", price=43.15) out = us._dedup([a, b, c, d]) assert len(out) == 3, [(x["src"], x["value"], x["dup"]) for x in out] assert out[0]["dup"] == 2 and out[1]["dup"] == 1 and out[2]["src"] == "decision" @case("单源失败不拖垮整体且逐源标明; 缓存按票生效; 实况分不编时间戳") def test_resilience_and_cache(): ymd = time.strftime("%Y-%m-%d") good = _FakeRedis({"intraday_signals:%s" % ymd: [ ("1-0", {"ts_code": "601126.SH", "producer_id": "bionic_decider", "action": "BUY", "trigger_time": _today_ms(10, 4), "suggested_price": "43.15"})]}, boom={"intraday_alerts:%s" % ymd, "mtf:intraday:stream:metrics"}) hits = {"n": 0} def fake_c208(db): hits["n"] += 1 return good def fake_mr(_c): return {"watch": [{"ts_code": "601126.SH", "score": 88}], "avoid": []} us._BYCODE_CACHE.clear() with _Patch() as p: p(us, "_c208", fake_c208) p(us, "_c214", lambda: object()) p(us, "_read_mr", fake_mr) out = us.by_code("SH601126") n1 = hits["n"] out2 = us.by_code("601126.SH") # 归一后同一只票, 走缓存 n2 = hits["n"] out3 = us.by_code("601126.SH", force=True) assert out["ok"] and len(out["timeline"]) == 1, out assert out["sources"]["intraday"]["ok"] and out["sources"]["sell_actions"]["ok"] assert out["sources"]["alerts"]["ok"] is False and "error" in out["sources"]["alerts"], out["sources"] assert out["sources"]["metrics"]["ok"] is False and out["sources"]["mr"]["ok"] is True assert n2 == n1 and out2 is out, "同一只票八秒内必须命中缓存" assert out3 is not out, "force 必须绕开缓存" assert out["snapshot"]["mr"]["watch"]["score"] == 88 and out["snapshot"]["mr"]["avoid"] is None assert all(r.get("ts") for r in out["timeline"]), "时间线每行都要有真实时间戳" assert "mr" not in {r["src"] for r in out["timeline"]}, "实况分没有时间, 绝不许排进时间线" @case("回扫上限参数已登记 (页面能改) 且有范围") def test_param_registered(): assert "PMS_UPSTREAM_BYCODE_SCAN" in param_store.RUNTIME_EXTRA assert "PMS_UPSTREAM_BYCODE_SCAN" in param_store._RANGES lo, hi = param_store._RANGES["PMS_UPSTREAM_BYCODE_SCAN"] assert lo >= 100 and hi <= 50000 with _Patch() as p: p(param_store, "get_int", lambda k, d=0: 1) assert us._bycode_scan() >= 100, "下限兜底防把回扫调成零" @case("只读守卫: 按代码回扫这段不许出现任何写入或消费组调用") def test_readonly_guard(): src = io.open(us.__file__.replace(".pyc", ".py"), encoding="utf-8").read() seg = src[src.index("# ================================================================ 按代码回扫"):] for bad in ("xadd", "xack", "xgroup", "xreadgroup", "xdel", "xtrim", "delete(", ".set(", "zadd"): assert bad not in seg.lower().replace("_bycode_cache.pop", ""), bad assert "xrevrange" in seg and "zrevrange" not in seg @case("页面静态断言: 三处入口、抽屉层级压得住浮层与遮罩、不进轮询、失败态不说今天没有") def test_page_static(): from app.web import main as web html = io.open(os.path.join(web.STATIC_DIR, "index.html"), encoding="utf-8").read() assert "/api/upstream/signals/" in html, "页面没接单票信号接口" assert html.count("openCodeSignals(") >= 6, "信号栏、持仓表、消息栏三处入口都要接上" # 层级: 弹层库从 2000 起自增, 浮层写死 2010、遮罩 2005。抽屉不显式抬层就落在它们底下, # 屏幕上正好只剩一层灰 —— 这就是遮罩空白那个症状的另一张脸。 rail = int(re.search(r"\.rail\.open\{[^}]*z-index:\s*(\d+)", html).group(1)) scrim = int(re.search(r"\.rail-scrim\{[^}]*z-index:\s*(\d+)", html).group(1)) dz = int(re.search(r"\.cs-drawer\{[^}]*z-index:\s*(\d+)", html).group(1)) sz = int(re.search(r"\.cs-scrim\{[^}]*z-index:\s*(\d+)", html).group(1)) assert dz > rail and dz > scrim and sz > rail, (dz, sz, rail, scrim) assert "if (railOpen.value) closeRail();" in html, "打开抽屉前要先收浮层 (双保险的另一半)" # 只能点击触发: 回扫接口全页只出现一次, 且不在整页刷新与任何定时器里 assert html.count("'/api/upstream/signals/'") == 1 load_all = html[html.index("async function loadAll()"):][:1600] assert "csLoadFull" not in load_all and "openCodeSignals" not in load_all, "回扫绝不许进整页刷新" for fn in ("startAutoRefresh", "wsPollSync"): if fn in html: seg = html[html.index("function " + fn):][:1200] assert "csLoadFull" not in seg, fn + " 里不许调回扫" # 读不到时不许说"今天没有" assert html.count("srcFail(") >= 8, "各路空文案都要给失败态让位" assert "别当成平安无事" in html @case("路由清单收了新接口") def test_route_registered(): from app.web.main import app paths = {r.path for r in app.routes} assert "/api/upstream/signals/{ts_code}" in paths src = io.open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "test_wiring.py"), encoding="utf-8").read() assert "/api/upstream/signals/{ts_code}" in src, "test_wiring 的路由清单也要加, 否则漏了没人发现" 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())