from __future__ import annotations import uuid from datetime import datetime, timezone from typing import Any from investly.domain import ( Action, Bar, Fundamentals, Market, Recommendation, RuleCitation, Strategy, ) from investly.engine.targets import target_plan ENGINE_VERSION = "baseline-0.2.1" def _clip(value: float, low: float = 0.0, high: float = 10.0) -> float: return max(low, min(high, value)) def investment_score( features: dict[str, float | None], fundamentals: Fundamentals ) -> tuple[float, list[str], list[str], bool]: score = 5.0 positives: list[str] = [] negatives: list[str] = [] fundamental_fields = [ fundamentals.pe, fundamentals.pb, fundamentals.roe, fundamentals.debt_to_equity, fundamentals.revenue_growth, fundamentals.earnings_growth, ] data_ok = sum(value is not None for value in fundamental_fields) >= 2 if not data_ok: negatives.append("insufficient investment-grade fundamental evidence") trend = features.get("dist_sma200") if trend is not None: if trend > 0: score += 0.8 positives.append("price above 200-day trend") else: score -= 0.8 negatives.append("price below 200-day trend") if fundamentals.roe is not None: roe = fundamentals.roe / 100 if fundamentals.roe > 1 else fundamentals.roe if roe >= 0.15: score += 1.0 positives.append("strong return on equity") elif roe < 0.08: score -= 0.7 negatives.append("weak return on equity") if fundamentals.revenue_growth is not None: if fundamentals.revenue_growth > 0: score += 0.7 positives.append("positive revenue growth") else: score -= 0.7 negatives.append("negative revenue growth") if fundamentals.earnings_growth is not None: if fundamentals.earnings_growth > 0: score += 0.8 positives.append("positive earnings growth") else: score -= 0.8 negatives.append("negative earnings growth") if fundamentals.pe is not None: if 0 < fundamentals.pe <= 20: score += 0.6 positives.append("reasonable earnings multiple") elif fundamentals.pe > 40: score -= 0.6 negatives.append("high earnings multiple") vol = features.get("vol60") if vol is not None and vol > 0.55: score -= 0.5 negatives.append("high realized volatility") return _clip(score), positives, negatives, data_ok def swing_score(features: dict[str, float | None]) -> tuple[float, list[str], list[str]]: score = 5.0 positives: list[str] = [] negatives: list[str] = [] m3 = features.get("mom_3m") d50 = features.get("dist_sma50") d200 = features.get("dist_sma200") vol = features.get("vol60") if d50 is not None: if d50 > 0: score += 1.0 positives.append("above 50-day trend") else: score -= 1.0 negatives.append("below 50-day trend") if d200 is not None and d200 > 0: score += 0.7 positives.append("long trend supportive") if m3 is not None: if 0.02 <= m3 <= 0.30: score += 1.2 positives.append("positive 3-month momentum") elif m3 < -0.08: score -= 1.2 negatives.append("negative 3-month momentum") elif m3 > 0.45: score -= 0.5 negatives.append("momentum potentially extended") if vol is not None and vol > 0.65: score -= 0.6 negatives.append("very high volatility") return _clip(score), positives, negatives def contextualize_for_position(rec: Recommendation, *, is_held: bool) -> Recommendation: """Convert entry-oriented analysis into a lifecycle instruction for held positions. The core score remains unchanged. Context only changes the action semantics: WATCH becomes HOLD for an existing position, while AVOID becomes SELL. A fresh BUY remains BUY, allowing the recommendation engine to explicitly add to a position. """ if not is_held: return rec if rec.action == Action.WATCH: rec.action = Action.HOLD rec.rules.append( RuleCitation( f"STRAT.{rec.strategy.value.upper()}.HOLD.001", 1, "existing position remains acceptable but does not meet fresh BUY threshold", ) ) elif rec.action == Action.AVOID: rec.action = Action.SELL rec.rules.append( RuleCitation( f"STRAT.{rec.strategy.value.upper()}.EXIT.001", 1, "existing position no longer meets minimum engine quality threshold", ) ) return rec def make_recommendation( *, run_id: str, market: Market, symbol: str, name: str, strategy: Strategy, features: dict[str, float | None], bars: list[Bar], fundamentals: Fundamentals, sources: list[str], evidence_snapshot: dict[str, Any], decision_at: datetime | None = None, ) -> Recommendation: investment_data_ok = True if strategy == Strategy.INVESTMENT: score, positives, negatives, investment_data_ok = investment_score(features, fundamentals) else: score, positives, negatives = swing_score(features) price = float(features["price"] or 0) plan = target_plan(strategy, bars, price) non_null_features = sum(value is not None for value in features.values()) confidence = min( 0.9, 0.45 + min(non_null_features / 12, 0.25) + min(len(positives) * 0.04, 0.2), ) if strategy == Strategy.INVESTMENT and not investment_data_ok: confidence = min(confidence, 0.49) actionable_target = plan.target is not None and plan.target > price if score >= 7 and confidence >= 0.58 and actionable_target and investment_data_ok: action = Action.BUY elif score >= 5.5: action = Action.WATCH else: action = Action.AVOID rule = RuleCitation( f"STRAT.{strategy.value.upper()}.ENTRY.001", 3, f"baseline score={score:.2f}, confidence={confidence:.2f}; target_method={plan.target_method}", ) profile: list[str] = [] if fundamentals.roe is not None and ( fundamentals.roe / 100 if fundamentals.roe > 1 else fundamentals.roe ) >= 0.15: profile.append("Quality") if (features.get("mom_3m") or 0) > 0: profile.append("Momentum") return Recommendation( recommendation_id=str(uuid.uuid4()), run_id=run_id, market=market, symbol=symbol, name=name, strategy=strategy, action=action, score=round(score, 2), confidence=round(confidence, 2), reference_price=round(price, 4), target=round(plan.target, 4) if plan.target is not None else None, stop=round(plan.stop, 4) if plan.stop is not None else None, horizon="6–18 months" if strategy == Strategy.INVESTMENT else "days–12 weeks", target_method=plan.target_method, invalidation_method=plan.invalidation_method, profile=profile or ["Baseline"], reasons_for=positives, reasons_against=negatives, rules=[rule], data_sources=sorted(set(sources)), evidence_snapshot=evidence_snapshot, engine_version=ENGINE_VERSION, created_at=decision_at or datetime.now(timezone.utc), )