81 lines
3.4 KiB
Python
81 lines
3.4 KiB
Python
"""Review feedback weights for opportunity ranking."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from app.research.review_agent import build_research_review
|
|
|
|
|
|
async def build_ranking_feedback(days: int = 60) -> dict:
|
|
review = await build_research_review(days=days)
|
|
theme_weights = _weights_from_breakdown(review.get("theme_breakdown", []), positive=True)
|
|
chain_weights = _weights_from_breakdown(review.get("chain_breakdown", []), positive=True)
|
|
signal_weights = _weights_from_breakdown(review.get("signal_breakdown", []), positive=True)
|
|
risk_weights = _weights_from_breakdown(review.get("risk_breakdown", []), positive=False)
|
|
return {
|
|
"days": days,
|
|
"sample_count": review.get("sample_count", 0),
|
|
"tracked_count": review.get("tracked_count", 0),
|
|
"theme_weights": theme_weights,
|
|
"chain_weights": chain_weights,
|
|
"signal_weights": signal_weights,
|
|
"risk_weights": risk_weights,
|
|
"summary": review.get("summary", {}),
|
|
}
|
|
|
|
|
|
def apply_feedback_to_card(card: dict, rec, risk_alerts: list[dict], feedback: dict | None) -> dict:
|
|
if not feedback or int(feedback.get("tracked_count") or 0) < 5:
|
|
return {**card, "review_adjustment": 0, "adjusted_score": card.get("score", 0), "review_feedback": []}
|
|
|
|
adjustments: list[dict] = []
|
|
theme = str(card.get("theme") or "")
|
|
chain_node = str(card.get("chain_node") or "")
|
|
signal = str(getattr(rec, "entry_signal_type", "") or "")
|
|
|
|
_append_adjustment(adjustments, "主题", theme, feedback.get("theme_weights", {}).get(theme, 0))
|
|
_append_adjustment(adjustments, "环节", chain_node, feedback.get("chain_weights", {}).get(chain_node, 0))
|
|
_append_adjustment(adjustments, "信号", signal, feedback.get("signal_weights", {}).get(signal, 0))
|
|
|
|
ts_code = str(card.get("ts_code") or "")
|
|
for risk in risk_alerts:
|
|
if risk.get("ts_code") and risk.get("ts_code") != ts_code:
|
|
continue
|
|
risk_type = str(risk.get("risk_type") or "")
|
|
_append_adjustment(adjustments, "风险", risk_type, feedback.get("risk_weights", {}).get(risk_type, 0))
|
|
|
|
total = round(sum(item["delta"] for item in adjustments), 1)
|
|
base_score = float(card.get("score") or 0)
|
|
adjusted = round(max(0, min(100, base_score + total)), 1)
|
|
return {
|
|
**card,
|
|
"review_adjustment": total,
|
|
"adjusted_score": adjusted,
|
|
"review_feedback": adjustments[:5],
|
|
}
|
|
|
|
|
|
def _weights_from_breakdown(items: list[dict], positive: bool) -> dict[str, float]:
|
|
weights: dict[str, float] = {}
|
|
for item in items:
|
|
label = str(item.get("label") or "")
|
|
tracked = int(item.get("tracked_count") or 0)
|
|
if not label or tracked < 2:
|
|
continue
|
|
avg_return = float(item.get("avg_return") or 0)
|
|
win_rate = float(item.get("win_rate") or 0)
|
|
if positive:
|
|
raw = (win_rate - 50) / 12 + avg_return * 0.65
|
|
delta = max(-4.0, min(6.0, raw))
|
|
else:
|
|
raw = avg_return * 0.85 - max(0, 45 - win_rate) / 10
|
|
delta = max(-8.0, min(0.0, raw))
|
|
if abs(delta) >= 0.5:
|
|
weights[label] = round(delta, 1)
|
|
return weights
|
|
|
|
|
|
def _append_adjustment(adjustments: list[dict], source: str, label: str, delta: float) -> None:
|
|
if not label or not delta:
|
|
return
|
|
adjustments.append({"source": source, "label": label, "delta": round(float(delta), 1)})
|