zlt
/
akg-factor-bridge
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
akg-factor-bridge/test_company_quality.py

114 lines
7.1 KiB
Python
Raw Normal View History

公司质地接入：公司深度取数、逻辑状态第五路、判决收敛只降不升、卡与计划两键、候选单表列、复盘分组、逐票接口回两键、单测（台账 053） Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-09 12:05:03 +08:00
"""公司质地接入（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()