from __future__ import annotations from dataclasses import dataclass, asdict from decimal import Decimal, ROUND_HALF_UP from typing import Any from src.market.context import CoinMarketContext, MarketContextSnapshot SCORE_VERSION = "confluence_score.v1" SCORE_WEIGHTS: dict[str, Decimal] = { "trend": Decimal("0.21"), "momentum": Decimal("0.17"), "volatility": Decimal("0.11"), "liquidity": Decimal("0.16"), "derivatives": Decimal("0.10"), "sentiment": Decimal("0.04"), "event_risk": Decimal("0.06"), "portfolio": Decimal("0.05"), "regime": Decimal("0.10"), "risk": Decimal("0.05"), } def _q(value: Decimal, places: str = "0.01") -> Decimal: return value.quantize(Decimal(places), rounding=ROUND_HALF_UP) def _clamp(value: Decimal, lo: Decimal = Decimal("0"), hi: Decimal = Decimal("100")) -> Decimal: return max(lo, min(hi, value)) @dataclass(frozen=True) class CoinConfluenceScore: coin: str trend_score: Decimal momentum_score: Decimal volatility_score: Decimal liquidity_score: Decimal derivatives_score: Decimal sentiment_score: Decimal event_risk_score: Decimal portfolio_score: Decimal regime_score: Decimal risk_score: Decimal final_trade_score: Decimal final_score: Decimal recommendation: str blockers: tuple[str, ...] components: dict[str, str] score_version: str = SCORE_VERSION weights: dict[str, Decimal] | None = None live_order_allowed: bool = False mainnet_signed_action: bool = False def to_dict(self) -> dict[str, Any]: raw = asdict(self) raw["trend_score"] = str(self.trend_score) raw["momentum_score"] = str(self.momentum_score) raw["volatility_score"] = str(self.volatility_score) raw["liquidity_score"] = str(self.liquidity_score) raw["derivatives_score"] = str(self.derivatives_score) raw["sentiment_score"] = str(self.sentiment_score) raw["event_risk_score"] = str(self.event_risk_score) raw["portfolio_score"] = str(self.portfolio_score) raw["regime_score"] = str(self.regime_score) raw["risk_score"] = str(self.risk_score) raw["final_trade_score"] = str(self.final_trade_score) raw["final_score"] = str(self.final_score) raw["blockers"] = list(self.blockers) raw["weights"] = {key: str(value) for key, value in (self.weights or SCORE_WEIGHTS).items()} return raw @dataclass(frozen=True) class MarketConfluenceReport: scores: dict[str, CoinConfluenceScore] market_regime: str eligible_count: int score_version: str = SCORE_VERSION weights: dict[str, Decimal] | None = None live_order_allowed: bool = False mainnet_signed_action: bool = False def to_dict(self) -> dict[str, Any]: return { "score_version": self.score_version, "weights": {key: str(value) for key, value in (self.weights or SCORE_WEIGHTS).items()}, "market_regime": self.market_regime, "eligible_count": self.eligible_count, "live_order_allowed": self.live_order_allowed, "mainnet_signed_action": self.mainnet_signed_action, "scores": {coin: score.to_dict() for coin, score in self.scores.items()}, } def _score_risk(*, blockers: list[str], ctx: CoinMarketContext) -> Decimal: risk_score = Decimal("85") hard_penalties = { "stale_data": Decimal("25"), "low_reliability": Decimal("20"), "spread_unknown": Decimal("15"), "l2_unavailable": Decimal("15"), "volatility_too_high": Decimal("20"), "mark_oracle_basis_wide": Decimal("10"), } for blocker in dict.fromkeys(blockers): risk_score -= hard_penalties.get(blocker, Decimal("5")) if ctx.reliability_score < Decimal("0.90"): risk_score -= Decimal("5") return _clamp(risk_score) def score_coin_confluence(ctx: CoinMarketContext, *, btc_trend_positive: bool = True, min_final_score: Decimal = Decimal("60"), event_risk: dict[str, Any] | None = None) -> CoinConfluenceScore: blockers: list[str] = [] mid = ctx.mid if ctx.mid > 0 else Decimal("1") atr_pct = (ctx.atr / mid * Decimal("100")) if ctx.atr > 0 and mid > 0 else Decimal("0") trend_score = Decimal("50") if ctx.sma_fast > ctx.sma_slow: trend_score += Decimal("30") elif ctx.sma_fast < ctx.sma_slow: trend_score -= Decimal("25") if ctx.mid > ctx.sma_fast > 0: trend_score += Decimal("15") trend_score = _clamp(trend_score) momentum_score = Decimal("50") if Decimal("45") <= ctx.rsi <= Decimal("65"): momentum_score += Decimal("25") elif Decimal("65") < ctx.rsi <= Decimal("72"): momentum_score += Decimal("5") elif ctx.rsi > Decimal("72"): momentum_score -= Decimal("25") blockers.append("momentum_overheated") elif ctx.rsi < Decimal("40"): momentum_score -= Decimal("15") momentum_score = _clamp(momentum_score) volatility_score = Decimal("50") if Decimal("0.35") <= atr_pct <= Decimal("2.50"): volatility_score += Decimal("25") elif atr_pct > Decimal("4.00"): volatility_score -= Decimal("25") blockers.append("volatility_too_high") elif Decimal("0") < atr_pct < Decimal("0.20"): volatility_score -= Decimal("15") volatility_score = _clamp(volatility_score) liquidity_score = Decimal("50") liquidity_notional = ctx.day_ntl_vlm if ctx.day_ntl_vlm > 0 else ctx.volume if liquidity_notional >= Decimal("25000000"): liquidity_score += Decimal("25") elif liquidity_notional < Decimal("12000000"): liquidity_score -= Decimal("25") blockers.append("weak_liquidity") if ctx.spread_pct is None: liquidity_score -= Decimal("15") blockers.append("spread_unknown") elif ctx.spread_pct <= Decimal("0.08"): liquidity_score += Decimal("15") else: liquidity_score -= Decimal("20") blockers.append("spread_too_wide") if not ctx.l2_available: liquidity_score -= Decimal("15") blockers.append("l2_unavailable") if ctx.buy_impact_1k_pct is None or ctx.sell_impact_1k_pct is None: liquidity_score -= Decimal("10") blockers.append("impact_unknown") else: max_impact_1k = max(ctx.buy_impact_1k_pct, ctx.sell_impact_1k_pct) if max_impact_1k > Decimal("0.12"): liquidity_score -= Decimal("20") blockers.append("impact_slippage_too_high") elif max_impact_1k <= Decimal("0.03"): liquidity_score += Decimal("5") liquidity_score = _clamp(liquidity_score) mark_oracle_basis_pct = Decimal("0") if ctx.oracle_px > 0 and ctx.mark_px > 0: mark_oracle_basis_pct = (ctx.mark_px - ctx.oracle_px) / ctx.oracle_px * Decimal("100") derivatives_score = Decimal("60") if ctx.open_interest <= 0: derivatives_score -= Decimal("10") blockers.append("open_interest_unknown") if Decimal("-0.00005") <= ctx.funding <= Decimal("0.00005"): derivatives_score += Decimal("15") elif ctx.funding > Decimal("0.0001"): derivatives_score -= Decimal("30") blockers.append("crowded_positive_funding") elif ctx.funding < Decimal("-0.00025"): derivatives_score -= Decimal("15") blockers.append("stressed_negative_funding") if abs(ctx.premium) <= Decimal("0.0005"): derivatives_score += Decimal("10") elif ctx.premium > Decimal("0.001"): derivatives_score -= Decimal("25") blockers.append("premium_too_high") elif ctx.premium < Decimal("-0.002"): derivatives_score -= Decimal("10") blockers.append("premium_dislocated_negative") if abs(mark_oracle_basis_pct) > Decimal("0.20"): derivatives_score -= Decimal("10") blockers.append("mark_oracle_basis_wide") derivatives_score = _clamp(derivatives_score) # v1 placeholders are explicit neutral scores until external sources are wired. sentiment_score = Decimal("50") event_risk = event_risk or {} event_risk_score = Decimal(str(event_risk.get("score", "70.00"))) event_risk_level = str(event_risk.get("risk_level", "unknown")) if event_risk_level == "red" or event_risk_score < Decimal("40"): blockers.append("event_risk_red") elif event_risk_level == "yellow" or event_risk_score < Decimal("60"): blockers.append("event_risk_yellow") portfolio_score = Decimal("50") regime_score = Decimal("65") if btc_trend_positive else Decimal("35") if ctx.stale_data: regime_score -= Decimal("25") blockers.append("stale_data") if ctx.reliability_score < Decimal("0.75"): regime_score -= Decimal("20") blockers.append("low_reliability") regime_score = _clamp(regime_score) risk_score = _score_risk(blockers=blockers, ctx=ctx) if risk_score < Decimal("50"): blockers.append("risk_score_too_low") final_trade_score = ( trend_score * SCORE_WEIGHTS["trend"] + momentum_score * SCORE_WEIGHTS["momentum"] + volatility_score * SCORE_WEIGHTS["volatility"] + liquidity_score * SCORE_WEIGHTS["liquidity"] + derivatives_score * SCORE_WEIGHTS["derivatives"] + sentiment_score * SCORE_WEIGHTS["sentiment"] + event_risk_score * SCORE_WEIGHTS["event_risk"] + portfolio_score * SCORE_WEIGHTS["portfolio"] + regime_score * SCORE_WEIGHTS["regime"] + risk_score * SCORE_WEIGHTS["risk"] ) final_trade_score = _q(_clamp(final_trade_score)) if final_trade_score < min_final_score: blockers.append("confluence_below_threshold") blockers_tuple = tuple(dict.fromkeys(blockers)) recommendation = "paper_candidate" if final_trade_score >= min_final_score and not blockers_tuple else "block_new_entry" return CoinConfluenceScore( coin=ctx.coin.upper(), trend_score=_q(trend_score), momentum_score=_q(momentum_score), volatility_score=_q(volatility_score), liquidity_score=_q(liquidity_score), derivatives_score=_q(derivatives_score), sentiment_score=_q(sentiment_score), event_risk_score=_q(event_risk_score), portfolio_score=_q(portfolio_score), regime_score=_q(regime_score), risk_score=_q(risk_score), final_trade_score=final_trade_score, final_score=final_trade_score, recommendation=recommendation, blockers=blockers_tuple, components={ "rsi": str(_q(ctx.rsi)), "atr_pct": str(_q(atr_pct)), "spread_pct": str(_q(ctx.spread_pct)) if ctx.spread_pct is not None else "unknown", "volume": str(_q(ctx.volume, "0.01")), "liquidity_notional": str(_q(liquidity_notional, "0.01")), "buy_impact_1k_pct": str(_q(ctx.buy_impact_1k_pct)) if ctx.buy_impact_1k_pct is not None else "unknown", "sell_impact_1k_pct": str(_q(ctx.sell_impact_1k_pct)) if ctx.sell_impact_1k_pct is not None else "unknown", "buy_impact_5k_pct": str(_q(ctx.buy_impact_5k_pct)) if ctx.buy_impact_5k_pct is not None else "unknown", "sell_impact_5k_pct": str(_q(ctx.sell_impact_5k_pct)) if ctx.sell_impact_5k_pct is not None else "unknown", "funding": str(_q(ctx.funding, "0.00000001")), "premium": str(_q(ctx.premium, "0.00000001")), "open_interest": str(_q(ctx.open_interest, "0.01")), "mark_oracle_basis_pct": str(_q(mark_oracle_basis_pct)), "reliability": str(_q(ctx.reliability_score)), "sentiment_source": "neutral_pending_external_feed", "event_risk_level": event_risk_level, "event_risk_reasons": ",".join(str(x) for x in event_risk.get("reasons", [])) if isinstance(event_risk.get("reasons"), list) else "", "portfolio_source": "neutral_pending_portfolio_context", }, weights=SCORE_WEIGHTS, ) def build_market_confluence_report(snapshot: MarketContextSnapshot, *, min_final_score: Decimal = Decimal("60"), event_risk: dict[str, dict[str, Any]] | None = None) -> MarketConfluenceReport: btc = snapshot.contexts.get("BTC") btc_trend_positive = True if btc is None else btc.sma_fast >= btc.sma_slow event_risk = event_risk or {} scores = {coin: score_coin_confluence(ctx, btc_trend_positive=btc_trend_positive, min_final_score=min_final_score, event_risk=event_risk.get(coin)) for coin, ctx in sorted(snapshot.contexts.items())} eligible = sum(1 for score in scores.values() if score.recommendation == "paper_candidate") regime = "risk_on" if btc_trend_positive and eligible >= 2 else "selective" if eligible else "risk_off" return MarketConfluenceReport(scores=scores, market_regime=regime, eligible_count=eligible, weights=SCORE_WEIGHTS)