114 lines
7.1 KiB
Python
114 lines
7.1 KiB
Python
"""公司质地接入（2026-09-09 接入方案第四节）：取数、第五路信号与合成、判决收敛只降不升、卡上文字。离线，不连库。"""
|
||
import datetime as dt
|
||
import json
|
||
import sys
|
||
import types
|
||
|
||
for _n in ("pandas", "pymysql", "psycopg"):
|
||
if _n not in sys.modules:
|
||
try:
|
||
__import__(_n)
|
||
except Exception: # noqa: BLE001
|
||
_m = types.ModuleType(_n)
|
||
if _n == "pandas":
|
||
_m.DataFrame = type("DataFrame", (), {})
|
||
sys.modules[_n] = _m
|
||
|
||
import card # noqa: E402
|
||
import logic_state as ls # noqa: E402
|
||
import sources # noqa: E402
|
||
|
||
|
||
def t(name, cond):
|
||
print((" ok " if cond else " FAIL") + name)
|
||
assert cond, name
|
||
|
||
|
||
class _DF:
|
||
def __init__(self, rows): self._rows = rows
|
||
def to_dict(self, _kind): return self._rows
|
||
|
||
|
||
def _review(overall="好", doubt=False, ran="2026-09-08", groups=None):
|
||
return {"overall": overall, "groups": groups or {"回报与护城河": "好", "盈余质量与财务安全": "好", "成长与含金量": "好"},
|
||
"doubt_hard": doubt, "ran_at": ran, "age_days": 1, "period": "2026Q1",
|
||
"grade_counts": {"好": 5, "中": 3, "差": 1, "证据不足": 1}, "valuation": "贵",
|
||
"thesis": "质地好但估值贵", "confidence": "中", "invalidation": "下一期看现金流", "quality_reasons": ["回报类好"],
|
||
"report_url": "http://192.168.16.178:8000/company/601126.SH/review"}
|
||
|
||
|
||
def test_sources():
|
||
calls = []
|
||
def rp(sql, params=None):
|
||
calls.append(params)
|
||
return _DF([{"ts_code": "601126.SH", "period": "2026Q1", "review_version": "2026.09-2", "grades": json.dumps({"penman": "好"}),
|
||
"summary": json.dumps({"overall": "好", "groups": {"回报与护城河": "好"}, "valuation": "贵", "doubt_hard": False,
|
||
"thesis": "x", "confidence": "中", "grade_counts": {"好": 5, "中": 3, "差": 1, "证据不足": 1},
|
||
"report_path": "/company/601126.SH/review"}, ensure_ascii=False),
|
||
"brain_used": True, "brain_status": "ok", "ran_at": dt.datetime(2026, 9, 8, 10, 0)},
|
||
{"ts_code": "688008.SH", "period": "2026Q1", "review_version": "2026.09-2", "grades": {"buffett": "差"},
|
||
"summary": None, "brain_used": False, "brain_status": None, "ran_at": dt.datetime(2026, 5, 1, 10, 0)}])
|
||
out = sources.company_reviews(["SH601126", "SH688008", "SZ000001"], "2026-09-09", read_pg=rp)
|
||
t("按点后缀码查", calls[0][0] == ["601126.SH", "688008.SH", "000001.SZ"])
|
||
t("摘要解析与报告链接", out["SH601126"]["overall"] == "好" and out["SH601126"]["report_url"].endswith("/company/601126.SH/review"))
|
||
t("运行日距数据日", out["SH601126"]["age_days"] == 1 and out["SH688008"]["age_days"] == 131)
|
||
t("没摘要的老行只有评价计数", out["SH688008"]["overall"] is None and out["SH688008"]["grade_counts"]["差"] == 1)
|
||
t("没报告的票不在字典里", "SZ000001" not in out)
|
||
def boom(*a, **k): raise RuntimeError("no table")
|
||
t("表读不到整体缺席", sources.company_reviews(["SH601126"], "2026-09-09", read_pg=boom) == {})
|
||
|
||
|
||
def test_fifth_path():
|
||
up = ls.from_company(_review("好"), "2026-09-09")
|
||
t("质地好向好且带截止日", up["signal"] == ls.SIG_UP and up["as_of"] == "2026-09-08" and up["path"] == ls.PATH_COMPANY)
|
||
flat = ls.from_company(_review("中"), "2026-09-09")
|
||
t("质地中平稳", flat["signal"] == ls.SIG_FLAT)
|
||
down = ls.from_company(_review("差", doubt=True), "2026-09-09")
|
||
t("质地差且盈余差：转弱且达到存疑条件", down["signal"] == ls.SIG_DOWN and down["hard"])
|
||
soft = ls.from_company(_review("差", doubt=False), "2026-09-09")
|
||
t("质地差但盈余不差：转弱不硬", soft["signal"] == ls.SIG_DOWN and not soft["hard"])
|
||
t("没报告缺失", ls.from_company(None, "2026-09-09")["signal"] == ls.SIG_NONE)
|
||
stale = _review("好"); stale["age_days"] = 200
|
||
t("报告陈旧缺失", ls.from_company(stale, "2026-09-09")["signal"] == ls.SIG_NONE)
|
||
t("证据不足缺失", ls.from_company(_review("证据不足"), "2026-09-09")["signal"] == ls.SIG_NONE)
|
||
# 合成：质地好单独不能定态（慢信号）；配研报论断向好则强化；质地差硬转弱单独就存疑
|
||
alone = ls.compose([ls.signal(ls.PATH_CLAIM, ls.SIG_NONE, why="缺"), up])
|
||
t("只剩公司质地这一路慢信号不能单独定态", alone["state"] == ls.STATE_UNKNOWN)
|
||
with_claim = ls.compose([ls.signal(ls.PATH_CLAIM, ls.SIG_UP, as_of="2026-09-08", coverage=3, why="利好"), up])
|
||
t("研报论断向好加质地好为逻辑强化", with_claim["state"] == ls.STATE_STRONG)
|
||
doubt = ls.compose([ls.signal(ls.PATH_CLAIM, ls.SIG_UP, as_of="2026-09-08", coverage=3, why="利好"), down])
|
||
t("一路向好一路转弱是证据矛盾交人", doubt["state"] == ls.STATE_UNKNOWN)
|
||
doubt2 = ls.compose([ls.signal(ls.PATH_CLAIM, ls.SIG_NONE, why="缺"), down])
|
||
t("质地差达到进入条件则逻辑存疑", doubt2["state"] == ls.STATE_DOUBT)
|
||
t("存疑触发表里有第五路", ls.PATH_COMPANY in ls.DOUBT_TRIGGERS and ls.SPEED[ls.PATH_COMPANY] == "慢")
|
||
|
||
|
||
def test_card_gate_and_views():
|
||
ev = {"pointed": True, "theme": "端侧AI芯片", "n_sources": 3, "pct0": 4.0, "covered": True, "upside": 0.2, "risk_name": False,
|
||
"accum_state": "明确吸筹·量在价先", "accum_score": 80, "accum_age": 1, "y_signal": "WATCH", "stale_snapshot": False, "logic": []}
|
||
base = card.judge(ev)
|
||
t("底子是候选", base["verdict"] == "候选")
|
||
down = card.judge(ev, company_review=_review("差", doubt=True))
|
||
t("质地差候选降为关注且依据写明", down["verdict"] == "关注" and "公司质地评为差" in down["basis"] and down["company_gate"])
|
||
t("质地中不动", card.judge(ev, company_review=_review("中"))["verdict"] == "候选")
|
||
t("质地好不升档、证据不足不动", card.judge(ev, company_review=_review("好"))["verdict"] == "候选"
|
||
and card.judge(ev, company_review=_review("证据不足"))["verdict"] == "候选")
|
||
t("开关关掉不降", card.judge(ev, company_review=_review("差"), quality_gate=False)["verdict"] == "候选")
|
||
watch_ev = {**ev, "accum_state": ""}
|
||
t("本来就是关注的不再变", card.judge(watch_ev, company_review=_review("差"))["verdict"] == "关注")
|
||
v = card.company_review_view(_review("好"))
|
||
t("整句带质地三类、十视角计数、论点、失效条件、链接", all(x in v for x in ("质地好", "回报好", "好5/中3/差1", "买方论点", "失效条件", "报告 http")))
|
||
t("短句与分组标签", card.company_review_short(_review("好")).startswith("好·贵") and card.company_review_label(_review("证据不足")) == "无"
|
||
and card.company_review_label(None) == "无" and card.company_review_view(None).startswith("公司深度：暂无"))
|
||
|
||
|
||
def main():
|
||
test_sources()
|
||
test_fifth_path()
|
||
test_card_gate_and_views()
|
||
print("ALL OK — 公司质地：取数 / 第五路 / 判决收敛 / 卡上文字")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|