101 lines
4.5 KiB
Python
101 lines
4.5 KiB
Python
"""假设卡片五字段校验+向导 prompt(P4-1,决议 M).
|
|
|
|
- 白名单同源纪律(tickflow custom_signals_ai.py:34-49 范式):prompt 渲染与
|
|
normalize 校验共用 load_domains()=config/data_manifest.yaml domains
|
|
(与 sanguo_factor.data_gap_check 同一供给).
|
|
- 字段语义: QuantaAlpha prompts.yaml:19-27(单行防膨胀/logic 条件句式/
|
|
falsifiable=testability+证伪闭环,C5 双锚).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
|
|
import yaml
|
|
|
|
_WS = re.compile(r"\s+")
|
|
|
|
_SIGN_SYNONYMS = {
|
|
"positive": "positive", "正向": "positive", "看多": "positive",
|
|
"正面": "positive",
|
|
"negative": "negative", "负向": "negative", "看空": "negative",
|
|
"反面": "negative",
|
|
}
|
|
|
|
_REQUIRED = ("title", "logic", "expected_sign", "falsifiable", "data_needs")
|
|
|
|
_SYSTEM_TEMPLATE = """你是A股量化投研假设卡设计专家。用户会给出一句投资直觉/研究方向，你要把它结构化为一张假设卡片草稿（五字段 JSON），供研究者确认后进入因子流水线。你只做文字→结构，不产出任何数字结论——数字结论一律由评估管线计算。
|
|
|
|
可用数据域（data_needs 白名单，只能使用以下键，禁止自造）：
|
|
{domains}
|
|
|
|
输出字段：
|
|
- title: 假设陈述，一句话概括假设本身，不超过 80 字
|
|
- logic: 依据。必须用条件句式（If…, …; When…, …）写可迁移的因果逻辑，不超过 200 字
|
|
- expected_sign: 预期方向，只能是 "positive" 或 "negative"
|
|
- falsifiable: 可证伪声明。写明范围/条件/阈值使其可被数据检验（例如「若 X 事件后 N 日相对基准超额收益均值<=0 则假设证伪」），不超过 200 字
|
|
- data_needs: 关联数据域数组，从上面白名单中选，可为空数组 []
|
|
|
|
要求：
|
|
1. 只输出一个 JSON 对象，禁止 markdown 代码块、禁止任何解释或多余文字。
|
|
2. 每个字段必须单行，禁止换行。
|
|
3. 优先用最少的数据域表达；确需多个条件时 data_needs 逐个列出。
|
|
|
|
示例：
|
|
{{"title":"高管增持公告后 60 日超额收益为正","logic":"If 高管用真金白银增持且金额占持股市值比例高,则内部人信息优势预示基本面改善","expected_sign":"positive","falsifiable":"若增持公告后 60 日相对基准超额收益均值<=0,则假设证伪","data_needs":["corpus_sentiment"]}}"""
|
|
|
|
|
|
def load_domains() -> list[str]:
|
|
"""读 data_manifest.yaml domains 键清单;文件缺失/坏 YAML/形状不对
|
|
(合法 YAML 但顶层非 dict 或 domains 非 dict)一律 fail-soft 返回 []."""
|
|
path = os.environ.get("SANGUO_DATA_MANIFEST", "config/data_manifest.yaml")
|
|
try:
|
|
with open(path, encoding="utf-8") as f:
|
|
manifest = yaml.safe_load(f) or {}
|
|
except (OSError, yaml.YAMLError):
|
|
return []
|
|
if not isinstance(manifest, dict): # P2-4: 顶层 list/scalar 曾 500 穿透
|
|
return []
|
|
domains = manifest.get("domains") or {}
|
|
return sorted(domains.keys()) if isinstance(domains, dict) else []
|
|
|
|
|
|
def normalize_draft(raw: dict, domains: list[str]) -> dict:
|
|
"""LLM 草稿/用户提交共用的归一+校验(tickflow 同一验证器双点复用).
|
|
|
|
折行压单行/同义方向 coerce/白名单过滤/缺失与空值 raise ValueError(中文).
|
|
"""
|
|
out: dict = {}
|
|
for key in _REQUIRED:
|
|
val = raw.get(key)
|
|
if val is None:
|
|
raise ValueError(f"卡片缺字段: {key}")
|
|
out[key] = val
|
|
|
|
for key in ("title", "logic", "falsifiable"):
|
|
text = _WS.sub(" ", str(out[key])).strip()
|
|
if not text:
|
|
raise ValueError(f"卡片字段为空: {key}")
|
|
limit = 80 if key == "title" else 200
|
|
out[key] = text[:limit]
|
|
|
|
sign = str(out["expected_sign"]).strip()
|
|
if sign not in _SIGN_SYNONYMS:
|
|
raise ValueError(f"expected_sign 非法: {sign!r}(只认 positive/negative)")
|
|
out["expected_sign"] = _SIGN_SYNONYMS[sign]
|
|
|
|
needs = out["data_needs"]
|
|
if not isinstance(needs, list):
|
|
needs = [needs] if isinstance(needs, str) else []
|
|
whitelist = set(domains)
|
|
out["data_needs"] = [str(d) for d in needs if str(d) in whitelist]
|
|
return out
|
|
|
|
|
|
def build_wizard_messages(sentence: str, domains: list[str]) -> list[dict]:
|
|
"""向导两消息:system(白名单+要求+示例) + 原句 user 不包装."""
|
|
lines = "\n".join(f"- {d}" for d in domains) or "- (暂无已登记数据域,输出空数组)"
|
|
system = _SYSTEM_TEMPLATE.format(domains=lines)
|
|
return [{"role": "system", "content": system},
|
|
{"role": "user", "content": sentence}]
|