194 lines
7.4 KiB
Python
194 lines
7.4 KiB
Python
"""Opportunity card ranking agent."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from app.research.feedback_agent import apply_feedback_to_card
|
||
from app.research.industry_chain_agent import infer_chain_node
|
||
|
||
|
||
def build_opportunity_cards(
|
||
recommendations: list[Any],
|
||
stock_notes: list[dict],
|
||
risk_alerts: list[dict],
|
||
feedback: dict | None = None,
|
||
) -> list[dict]:
|
||
note_map = {item["ts_code"]: item for item in stock_notes}
|
||
rejected = {item["ts_code"] for item in risk_alerts if item.get("reject") and item.get("ts_code")}
|
||
risk_by_code: dict[str, list[dict]] = {}
|
||
global_risks: list[dict] = []
|
||
for risk in risk_alerts:
|
||
if risk.get("ts_code"):
|
||
risk_by_code.setdefault(risk["ts_code"], []).append(risk)
|
||
else:
|
||
global_risks.append(risk)
|
||
cards: list[dict] = []
|
||
for rec in recommendations:
|
||
ts_code = getattr(rec, "ts_code", "")
|
||
if ts_code in rejected:
|
||
continue
|
||
note = note_map.get(ts_code, {})
|
||
action_plan = getattr(rec, "action_plan", "观察") or "观察"
|
||
if action_plan == "可操作":
|
||
opportunity_type = "可操作"
|
||
elif action_plan == "重点关注":
|
||
opportunity_type = "等确认"
|
||
else:
|
||
hint = ((getattr(rec, "decision_trace", {}) or {}).get("position_adjustment") or {}).get("hint", "")
|
||
opportunity_type = "等回踩" if hint == "wait_pullback" else "仅观察"
|
||
theme = getattr(rec, "sector", "") or note.get("theme") or "未归类"
|
||
alpha_profile = _build_alpha_profile(rec, note, risk_by_code.get(ts_code, []) + global_risks)
|
||
card = {
|
||
"ts_code": ts_code,
|
||
"name": getattr(rec, "name", ""),
|
||
"theme": theme,
|
||
"chain_node": note.get("chain_node") or infer_chain_node(theme, getattr(rec, "name", ""), theme),
|
||
"stock_role": note.get("stock_role", "待归类"),
|
||
"opportunity_type": opportunity_type,
|
||
"score": round(float(getattr(rec, "score", 0) or 0), 1),
|
||
"logic_score": note.get("logic_score", 0),
|
||
"action_plan": action_plan,
|
||
"trigger": getattr(rec, "trigger_condition", "") or getattr(rec, "entry_timing", ""),
|
||
"invalid_condition": note.get("invalid_condition") or getattr(rec, "invalidation_condition", "") or getattr(rec, "risk_note", ""),
|
||
"logic_summary": note.get("logic_summary", ""),
|
||
**alpha_profile,
|
||
}
|
||
cards.append(apply_feedback_to_card(card, rec, risk_by_code.get(ts_code, []) + global_risks, feedback))
|
||
cards.sort(
|
||
key=lambda item: (
|
||
{"可操作": 3, "等确认": 2, "等回踩": 1, "仅观察": 0}.get(item["opportunity_type"], 0),
|
||
item.get("alpha_score", 0),
|
||
item.get("ambush_score", 0),
|
||
item.get("adjusted_score", item["score"]),
|
||
item["logic_score"],
|
||
),
|
||
reverse=True,
|
||
)
|
||
return cards[:20]
|
||
|
||
|
||
def _build_alpha_profile(rec: Any, note: dict, risks: list[dict]) -> dict:
|
||
trace = getattr(rec, "decision_trace", {}) or {}
|
||
position = trace.get("position_adjustment") or {}
|
||
hint = position.get("hint", "")
|
||
action_plan = getattr(rec, "action_plan", "观察") or "观察"
|
||
score = float(getattr(rec, "score", 0) or 0)
|
||
market_score = float(getattr(rec, "market_temp_score", 0) or 0)
|
||
sector_score = float(getattr(rec, "sector_score", 0) or 0)
|
||
capital_score = float(getattr(rec, "capital_score", 0) or 0)
|
||
technical_score = float(getattr(rec, "technical_score", 0) or 0)
|
||
valuation_score = float(getattr(rec, "valuation_score", 0) or 0)
|
||
position_score = float(getattr(rec, "position_score", 0) or 0)
|
||
logic_score = float(note.get("logic_score", 0) or 0)
|
||
risk_penalty = min(sum(18 if item.get("reject") else 7 for item in risks), 35)
|
||
|
||
beta_dependency_score = _clamp(
|
||
20
|
||
+ market_score * 0.55
|
||
+ max(0, sector_score - capital_score) * 0.2
|
||
- min(capital_score, 20) * 0.15
|
||
)
|
||
alpha_score = _clamp(
|
||
score * 0.25
|
||
+ sector_score * 0.18
|
||
+ capital_score * 0.22
|
||
+ technical_score * 0.16
|
||
+ logic_score * 0.14
|
||
+ valuation_score * 0.08
|
||
- beta_dependency_score * 0.12
|
||
- risk_penalty
|
||
)
|
||
ambush_score = _clamp(
|
||
position_score * 0.26
|
||
+ valuation_score * 0.22
|
||
+ capital_score * 0.18
|
||
+ technical_score * 0.16
|
||
+ (12 if hint in {"wait_confirm", "actionable_pullback", "wait_pullback"} else 0)
|
||
+ (8 if action_plan in {"重点关注", "观察"} else 0)
|
||
- risk_penalty * 0.7
|
||
)
|
||
expectation_gap_score = _clamp(
|
||
logic_score * 0.28
|
||
+ sector_score * 0.2
|
||
+ capital_score * 0.2
|
||
+ valuation_score * 0.16
|
||
+ (10 if note.get("generated_by") == "llm" else 4)
|
||
- market_score * 0.1
|
||
- risk_penalty * 0.6
|
||
)
|
||
|
||
risk_gate = _risk_gate(risks)
|
||
alpha_type = _alpha_type(action_plan, hint, ambush_score, expectation_gap_score, beta_dependency_score)
|
||
beta_dependency = _beta_label(beta_dependency_score)
|
||
setup_quality = _setup_quality(alpha_score, ambush_score, risk_gate)
|
||
|
||
return {
|
||
"alpha_type": alpha_type,
|
||
"alpha_score": round(alpha_score, 1),
|
||
"beta_dependency": beta_dependency,
|
||
"beta_dependency_score": round(beta_dependency_score, 1),
|
||
"ambush_score": round(ambush_score, 1),
|
||
"expectation_gap_score": round(expectation_gap_score, 1),
|
||
"risk_gate": risk_gate,
|
||
"setup_quality": setup_quality,
|
||
"alpha_reason": _alpha_reason(alpha_type, beta_dependency, ambush_score, expectation_gap_score, risk_gate),
|
||
}
|
||
|
||
|
||
def _alpha_type(action_plan: str, hint: str, ambush_score: float, expectation_gap_score: float, beta_dependency_score: float) -> str:
|
||
if ambush_score >= 70 and action_plan != "可操作":
|
||
return "低位埋伏"
|
||
if hint == "wait_pullback":
|
||
return "强势等回踩"
|
||
if hint == "wait_confirm" or action_plan == "重点关注":
|
||
return "等确认"
|
||
if expectation_gap_score >= 72 and beta_dependency_score <= 65:
|
||
return "预期差机会"
|
||
if action_plan == "可操作":
|
||
return "趋势确认"
|
||
return "观察线索"
|
||
|
||
|
||
def _risk_gate(risks: list[dict]) -> str:
|
||
if any(item.get("reject") for item in risks):
|
||
return "否决"
|
||
if risks:
|
||
return "预警"
|
||
return "通过"
|
||
|
||
|
||
def _beta_label(score: float) -> str:
|
||
if score >= 70:
|
||
return "高"
|
||
if score >= 45:
|
||
return "中"
|
||
return "低"
|
||
|
||
|
||
def _setup_quality(alpha_score: float, ambush_score: float, risk_gate: str) -> str:
|
||
if risk_gate == "否决":
|
||
return "不参与"
|
||
if alpha_score >= 78 and ambush_score >= 65:
|
||
return "优先研究"
|
||
if alpha_score >= 65 or ambush_score >= 70:
|
||
return "可跟踪"
|
||
return "仅观察"
|
||
|
||
|
||
def _alpha_reason(alpha_type: str, beta_dependency: str, ambush_score: float, expectation_gap_score: float, risk_gate: str) -> str:
|
||
if risk_gate == "否决":
|
||
return "风险门槛未通过，先排除。"
|
||
parts = [f"{alpha_type}，Beta依赖{beta_dependency}"]
|
||
if ambush_score >= 70:
|
||
parts.append("位置更适合提前跟踪")
|
||
if expectation_gap_score >= 70:
|
||
parts.append("存在预期差")
|
||
if risk_gate == "预警":
|
||
parts.append("但有风险预警")
|
||
return "；".join(parts) + "。"
|
||
|
||
|
||
def _clamp(value: float, low: float = 0, high: float = 100) -> float:
|
||
return max(low, min(high, value))
|