feat(prompt): 技术位锚定+止损门禁+12维命名统一

- collect_data: 现算technical_analysis.full_analysis(强撑/弱撑/枢轴/弱压/强压/有效区间/MA)
- prompt: 【技术位锚】独立高亮段+锚定纪律(止损<区下沿,锚弱撑下方1-3%/强撑;止盈锚弱压强压;
  偏离须说明理由)+输出纪律第4条+【参数自检】节
- save_result: sl>=el时用弱撑x0.985/强撑x0.99自动修正并留痕,无技术位拒写
- 命名: 九维→12维全局统一(prompt/docstring/前端弹窗)
This commit is contained in:
hmo
2026-07-22 23:43:45 +08:00
parent 440ce56b0b
commit cf039b2770
3 changed files with 79 additions and 18 deletions
+74 -13
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""batch_reassess.py — 批量补全12维(九维矩阵)LLM分析(逐只处理,间隔防限流)
"""batch_reassess.py — 批量补全12维矩阵LLM分析(逐只处理,间隔防限流)
用法:
python3 batch_reassess.py # 所有缺分析/过期的 active 策略
@@ -149,11 +149,35 @@ def collect_data(code):
conn.close()
except:
data["macro"] = "大盘震荡"
# ── 确定性技术位(2026-07-22 老爸:系统计算的技术位必须作为客观锚喂给LLM)──
# 与技术路径(per_stock)同源: technical_analysis.full_analysis
# 强撑/弱撑/枢轴/弱压/强压/有效区间 + 均线,全部确定性计算,非LLM估计。
data["ta"] = {}
try:
import technical_analysis as ta_mod
_ta = ta_mod.full_analysis(code)
if _ta and "error" not in _ta:
_sr = _ta.get("support_resistance", {}) or {}
_mtf = _ta.get("multi_tf", {}) or {}
_mas = (_mtf.get("mas") or {})
data["ta"] = {
"strong_support": _sr.get("strong_support"),
"weak_support": _sr.get("weak_support"),
"pivot": _sr.get("pivot"),
"weak_resist": _sr.get("weak_resist"),
"strong_resist": _sr.get("strong_resist"),
"effective_range": _sr.get("effective_range"),
"ma5": _mas.get("ma5"), "ma10": _mas.get("ma10"),
"ma20": _mas.get("ma20"), "ma60": _mas.get("ma60"),
}
except Exception as _te:
print(f" ⚠️ 技术位计算失败({code}): {_te}", flush=True)
return data
def build_prompt(data):
"""构建LLM prompt,先审阅原策略再结合实时数据输出修改判断+维矩阵分析"""
"""构建LLM prompt,先审阅原策略再结合实时数据输出修改判断+12维矩阵分析"""
cash, total = get_portfolio()
if not total:
cash, total = 241330, 929727 # 兜底(DB读不到时)
@@ -270,16 +294,33 @@ def build_prompt(data):
完整分析原文:
{_fa_display}"""
return f"""你是一个资深A股分析师。请先审阅以下【原策略全文】,判断是否需要修改策略,然后做出完整的九维矩阵分析。
# ── 技术位锚 section2026-07-22 老爸:确定性计算值,LLM 必须尊重)──
_ta = data.get("ta") or {}
if _ta.get("weak_support"):
_ta_sec = f"""【技术位锚】(以下数值由系统基于K线/均线确定性计算,是客观事实,不是你的估计值)
强撑={_ta.get('strong_support')} 弱撑={_ta.get('weak_support')} 枢轴={_ta.get('pivot')}
弱压={_ta.get('weak_resist')} 强压={_ta.get('strong_resist')} 有效区间={_ta.get('effective_range')}
MA5={_ta.get('ma5')} MA10={_ta.get('ma10')} MA20={_ta.get('ma20')} MA60={_ta.get('ma60')}
⚠️ 参数锚定纪律(必须遵守,输出前自检):
1. 买入区应落在技术位之间:下沿参考弱撑/强撑附近,上沿参考枢轴/弱压附近
2. 止损必须严格低于买入区下沿——放在弱撑下方1-3%或强撑附近;严禁止损≥区间下沿(等于把止损设在买入价上,下沿买入立即止损,RR恒为0)
3. 止盈应参考弱压/强压,不得明显高于强压
4. 若你判断技术位不适用(如突发重大消息/基本面剧变),必须在【修改点及理由】中明确写出偏离理由,禁止静默偏离"""
else:
_ta_sec = "【技术位锚】本次计算不可用,请基于价格行为谨慎给出参数,并仍须满足:止损<区间下沿<区间上沿<止盈。"
return f"""你是一个资深A股分析师。请先审阅以下【原策略全文】,判断是否需要修改策略,然后做出完整的12维矩阵分析。
【原策略全文】
{_orig_strategy_section}
── 以上是已有的策略,以下是当前实时数据,请结合两者做出判断 ──
⚠️ 重要:以下9个维度不是独立分析的,你必须交叉对比后给出综合结论。
⚠️ 重要:以下12个维度不是独立分析的,你必须交叉对比后给出综合结论。
例如:如果消息面利好但资金流在流出,说明利好可能是出货;如果基本面强但技术面破位,说明估值可能还没到底。
{_ta_sec}
当前数据(以下数据均来自实时API,每条标注时间窗口,禁止使用模型内部训练数据):
大盘:{data.get('macro','震荡')}(当日实时)
最新价:{data.get('price',0)} 涨跌:{data.get('change_pct','0')}%(当日实时)
@@ -319,9 +360,11 @@ PE={data.get('pe','?')}(最新财报) 市值={data.get('mcap','?')}亿
【综合结论】(买入/关注/观望/卖出)
【操作建议】具体操作建议
【买入区间】最低价~最高价
【建议止损】数字
【建议止盈】数字
【买入区间】最低价~最高价(锚定技术位:下沿参考弱撑/强撑,上沿参考枢轴/弱压)
【建议止损】数字(必须严格低于买入区下沿,放弱撑下方1-3%或强撑附近)
【建议止盈】数字(参考弱压/强压)
【参数自检】一行,格式"止损X < 区下沿Y < 区上沿Z < 止盈W:通过/不通过+原因"
【建议仓位】⚠️不可省略。综合结论非"买入"时写"不新建仓";为"买入"时按以下公式:
基础仓位按RR确定:RR<1.5→不推荐,RR1.5~3→8%RR3~5→12%RR5+→15%
@@ -334,7 +377,8 @@ PE={data.get('pe','?')}(最新财报) 市值={data.get('mcap','?')}亿
⚠️ 输出纪律(必须遵守):
1. 直接以【维持或修改】开头,禁止任何寒暄、开场白、分隔线
2. 禁止输出 <structured_data> 或任何 XML/JSON/代码块
3. 所有【】节标题一个都不能少"""
3. 所有【】节标题一个都不能少
4. 止损<区间下沿<区间上沿<止盈,违反任一条=输出作废重想"""
def parse_response(text):
"""从LLM回复中提取策略参数。
⚠️ 节标题精确匹配:只认行首【买入区间】【综合结论】等节行。
@@ -406,8 +450,9 @@ def parse_response(text):
return result
def save_result(code, full_text, parsed):
"""保存LLM结果到DB(先快照再UPDATE)。空分析拒绝写入。"""
def save_result(code, full_text, parsed, ta_levels=None):
"""保存LLM结果到DB(先快照再UPDATE)。空分析拒绝写入。
ta_levels: collect_data 计算的确定性技术位,用于止损锚定校验。"""
if not (full_text or "").strip():
print(f" \u274c 拒绝写入空分析(LLM输出为空,保护已有数据)")
return
@@ -452,6 +497,22 @@ def save_result(code, full_text, parsed):
print(f" ⚠️ 买入区解析异常({_el}~{_eh}),跳过区间写入(保留原值)", flush=True)
# 止损/止盈一致性门禁:损>0 时必须在区间下沿之下(0.5x~1.0x),盈>0 时必须在区间上沿之上
_sl, _tp = parsed["stop_loss"], parsed["take_profit"]
# ── 止损锚定门禁(2026-07-22 老爸:止损=区间下沿=把止损设在买入价上,RR恒0)──
# LLM 输出 sl>=el 时:用确定性技术位自动修正(弱撑×0.985),无技术位则拒写止损。
if _sl > 0 and _el > 0 and _sl >= _el:
_ws = (ta_levels or {}).get("weak_support") or 0
_ss = (ta_levels or {}).get("strong_support") or 0
if _ws > 0 and _ws < _el:
_fixed = round(_ws * 0.985, 2)
print(f" ⚠️ 止损{_sl}≥区下沿{_el},按技术锚修正为 弱撑{_ws}×0.985={_fixed}", flush=True)
_sl = _fixed
elif _ss > 0 and _ss < _el:
_fixed = round(_ss * 0.99, 2)
print(f" ⚠️ 止损{_sl}≥区下沿{_el},按技术锚修正为 强撑{_ss}×0.99={_fixed}", flush=True)
_sl = _fixed
else:
print(f" ⚠️ 止损{_sl}≥区下沿{_el}且无可用技术位,拒写止损(保留原值)", flush=True)
_sl = 0
if _sl > 0 and (not _el or _sl < _el) and (not _tp or _sl < _tp):
updates.append("stop_loss=?")
params.append(_sl)
@@ -507,7 +568,7 @@ def process_stock(code, force_today=False):
print(f" \u26a0\ufe0f 无价格数据,跳过")
return False
print(f" 调LLM生成维分析...", flush=True)
print(f" 调LLM生成12维分析...", flush=True)
prompt = build_prompt(data)
# ── 使用共享 LLM 客户端(替代 curl subprocess)──
@@ -533,7 +594,7 @@ def process_stock(code, force_today=False):
print(f" 信号={parsed['signal']} 区间={parsed['entry_low']}~{parsed['entry_high']} 损={parsed['stop_loss']} 盈={parsed['take_profit']} 仓位={parsed['position']}")
save_result(code, full_text, parsed)
save_result(code, full_text, parsed, ta_levels=data.get("ta"))
print(f" \u2705 已保存到DB")
return True
+3 -3
View File
@@ -42,7 +42,7 @@ from mofin_db import snapshot_strategy_history
def _build_full_analysis(code, entry, result):
"""从重评结果构建完整维分析文本"""
"""从重评结果构建完整12维分析文本"""
if not result:
return ""
lines = []
@@ -144,7 +144,7 @@ def _build_full_analysis(code, entry, result):
if ma_match:
ma5, ma10, ma20, ma60 = ma_match.groups()
lines.append(f"{name}({code} 维全析)】")
lines.append(f"{name}({code} 12维全析)】")
lines.append("")
if macro_desc:
lines.append(f"① 大盘环境(当日实时):{macro_desc}")
@@ -390,7 +390,7 @@ def main():
if _fa and _fa[0]: print(f" ✅ full_analysis已写入({len(_fa[0])}字)")
else: print(f" ⚠️ full_analysis为空")
_v.close()
# LLM生成完整维分析
# LLM生成完整12维分析
_macro_desc = ""
_pe_val = ""
_pb_val = ""