Spaces:
Running
Running
File size: 3,353 Bytes
48bd848 89afbf3 48bd848 89afbf3 48bd848 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | from __future__ import annotations
from app.scoring.explanations import build_score_explanation, opportunity_label, watch_points_from_factors
from app.scoring.normalization import avg, clamp, safe_float, scale
from app.scoring.weights import OPPORTUNITY_WEIGHTS, normalized_weights
def compute_opportunity_score(asset: dict, signal: dict | None, market_snapshot: dict, sector_score: float = 50, macro_score: float = 50, weights_override: dict | None = None) -> dict:
breakdown = (signal or {}).get("score_breakdown") or {}
narrative = (signal or {}).get("narrative_flags") or {}
technical = (signal or {}).get("technical_flags") or {}
momentum = avg(
safe_float(breakdown.get("momentum_score")),
scale(market_snapshot.get("perf_5d"), -8, 8),
scale(market_snapshot.get("perf_1m"), -14, 16),
)
trend = avg(safe_float(breakdown.get("trend_score")), 82 if technical.get("above_sma20") else 42, 82 if technical.get("above_sma50") else 42)
relative_strength = safe_float(breakdown.get("etf_confirmation_score"), 50)
volume = avg(scale(technical.get("volume_spike"), 0, 180), 72 if safe_float(market_snapshot.get("volume")) > 0 else 38)
sentiment = avg(safe_float(breakdown.get("sentiment_score")), scale(narrative.get("sentiment_7d"), -0.6, 0.6))
news = avg(scale(narrative.get("news_count_7d"), 0, 6), scale(narrative.get("narrative_intensity"), 0, 100))
risk = compute_risk_score(signal, technical)
weights = normalized_weights(weights_override or OPPORTUNITY_WEIGHTS)
opportunity = (
momentum * weights["momentum"]
+ trend * weights["trend"]
+ relative_strength * weights["relative_strength"]
+ volume * weights["volume"]
+ sentiment * weights["sentiment"]
+ news * weights["news"]
+ sector_score * weights["sector"]
+ macro_score * weights["macro"]
+ (100 - risk) * weights["risk"]
)
factors = {
"opportunity_score": round(clamp(opportunity), 1),
"trend_score": round(clamp(trend), 1),
"momentum_score": round(clamp(momentum), 1),
"sentiment_score": round(clamp(sentiment), 1),
"news_score": round(clamp(news), 1),
"risk_score": round(clamp(risk), 1),
"relative_strength_score": round(clamp(relative_strength), 1),
"volume_score": round(clamp(volume), 1),
"sector_score": round(clamp(sector_score), 1),
"macro_score": round(clamp(macro_score), 1),
}
ticker = asset.get("ticker", "")
return {
**factors,
"status_label": opportunity_label(factors["opportunity_score"], factors["risk_score"]),
"why_today": build_score_explanation(ticker, factors, factors["opportunity_score"]),
"watch_points": watch_points_from_factors(factors),
"weights": weights,
}
def compute_risk_score(signal: dict | None, technical: dict) -> float:
if signal and signal.get("risk_level") == "High":
base = 78
elif signal and signal.get("risk_level") == "Low":
base = 35
else:
base = 52
volatility = scale(technical.get("historical_volatility"), 12, 70)
drawdown = scale(abs(safe_float(technical.get("recent_drawdown"))), 0, 30)
overextension = scale(technical.get("rsi"), 58, 78)
return avg(base, volatility, drawdown, overextension)
|