"""Orchestrator — runs the full agent pipeline and persists results. Pipeline: Market Scanner → per-candidate: Volatility → Technical → Options → Sentiment → AI Probability → Risk Manager → Signal record (approved/rejected). """ import logging import os from sqlalchemy.orm import Session from .. import models from ..agents.market_scanner import MarketScannerAgent from ..agents.news import NewsAgent from ..agents.regime import REGIME_SYMBOL, RegimeAgent from ..agents.options import OptionsAnalysisAgent from ..agents.probability import ProbabilityAgent from ..agents.risk_manager import RiskManagerAgent from ..agents.sentiment import SentimentAgent from ..agents.swing import MIN_SWING_SCORE, STOP_ATR_MULT, SwingTradingAgent from ..agents.technical import TechnicalAnalysisAgent from ..agents.volatility import VolatilityAgent from ..services.earnings import days_to_next_earnings, earnings_guard_enabled from ..services.learning import get_calibration from ..services.market_data import bars_are_fresh, get_provider logger = logging.getLogger(__name__) def run_scan_pipeline(db: Session, max_candidates: int = 15) -> models.ScanRun: provider = get_provider() scanner = MarketScannerAgent(provider, max_candidates=max_candidates) vol_agent = VolatilityAgent() tech_agent = TechnicalAnalysisAgent() opt_agent = OptionsAnalysisAgent(provider) sent_agent = SentimentAgent() prob_agent = ProbabilityAgent() risk_agent = RiskManagerAgent() swing_agent = SwingTradingAgent() news_agent = NewsAgent() scan_result = scanner.run({}) rows = scan_result.data["rows"] candidates = scan_result.data["candidates"] run = models.ScanRun(universe_size=len(rows), candidates=len(candidates), results=rows) db.add(run) db.flush() open_risk_usd = _current_open_risk(db) calibration = get_calibration(db) # learning loop: past outcomes inform new estimates created = 0 stale_skipped = 0 # Market regime: classify once per scan from the index proxy regime_result = RegimeAgent().run({"history": provider.get_history(REGIME_SYMBOL, 90)}) regime = regime_result.data["regime"] risk_scale = regime_result.data["risk_scale"] regime_note = (f"Market regime: {regime.upper()}" + (" — high volatility, sizing halved" if regime_result.data["high_volatility"] else "")) logger.info(regime_note) guard_earnings = earnings_guard_enabled() for cand in candidates: symbol = cand["symbol"] # 90 calendar days ≈ 64 trading bars — enough for SMA50 (a 60-day # window yields only ~43 bars, silently disabling 50-bar indicators) history = provider.get_history(symbol, days=90) # Stale-data circuit breaker (audit Phase 5 / finding B-3): # old data must never generate actionable signals. if not bars_are_fresh(history): stale_skipped += 1 logger.warning("CIRCUIT BREAKER: %s data is stale — no signal generated", symbol) continue quote = provider.get_quote(symbol) ctx = {"symbol": symbol, "history": history, "quote": quote} days_to_earnings = days_to_next_earnings(symbol) if guard_earnings else None vol = vol_agent.run(ctx) tech = tech_agent.run(ctx) direction = tech.direction or "CALL" ctx["direction"] = direction opts = opt_agent.run(ctx) sent = sent_agent.run(ctx) news = news_agent.run(ctx) prob = prob_agent.run({**ctx, "calibration": calibration, "agent_results": { "volatility": vol, "technical": tech, "options": opts, "sentiment": sent, "news": news, }}) risk = risk_agent.run({ **ctx, "atr": vol.data.get("atr"), "confidence": prob.data["confidence"], "probability": prob.data["probability"], "open_risk_usd": open_risk_usd, "risk_scale": risk_scale, "days_to_earnings": days_to_earnings, }) rd = risk.data if rd["approved"]: open_risk_usd += rd["max_risk_usd"] signal = models.Signal( scan_run_id=run.id, symbol=symbol, company=cand["company"], signal_type="options", direction=direction, status="approved" if rd["approved"] else "rejected", risk_rating=rd["risk_rating"], price=quote["price"], confidence=prob.data["confidence"], probability=prob.data["probability"], entry=rd["entry"], stop_loss=rd["stop_loss"], target1=rd["target1"], target2=rd["target2"], risk_reward=rd["risk_reward"], position_size=rd["position_size"], max_risk_usd=rd["max_risk_usd"], scores={ "volatility": vol.score, "technical": tech.score, "options": opts.score, "sentiment": sent.score, "news": news.score, "confidence": prob.data["confidence"], }, rationale=[regime_note] + tech.notes + vol.notes + opts.notes + sent.notes + news.notes + prob.notes, rejection_reasons=rd["rejection_reasons"], option_contract=opts.data.get("contract", {}), indicators=tech.data.get("indicators", {}), history=[b["close"] for b in history[-30:]], ) db.add(signal) created += 1 # ── Swing-trade stock signal (shares, multi-day hold) ── # Regime gate: trend-following swing entries are disabled in chop swing = swing_agent.run(ctx) if regime != "chop" else None if swing and swing.direction and swing.score >= MIN_SWING_SCORE: # Ensemble: swing setup quality dominates; technical conviction # confirms; sentiment agreement nudges. tech_confirms = (tech.direction == "CALL") == (swing.direction == "BUY") swing_conf = swing.score * 0.55 + tech.score * (0.30 if tech_confirms else 0.10) \ + vol.score * 0.15 swing_conf = round(max(0.0, min(98.0, swing_conf)), 1) swing_prob = round(max(5.0, min(90.0, 23 + swing_conf * 0.5)), 1) if calibration and calibration.get("active"): from ..services.learning import adjust_probability swing_prob = adjust_probability(swing_prob, calibration) swing_risk = risk_agent.run({ **ctx, "direction": swing.direction, "atr": swing.data.get("atr"), "atr_mult": STOP_ATR_MULT, "confidence": swing_conf, "probability": swing_prob, "open_risk_usd": open_risk_usd, "risk_scale": risk_scale, "days_to_earnings": days_to_earnings, }) srd = swing_risk.data if srd["approved"]: open_risk_usd += srd["max_risk_usd"] swing_notes = swing.notes + ( ["Technical agent confirms the swing direction"] if tech_confirms else ["Technical agent disagrees — confidence reduced"] ) db.add(models.Signal( scan_run_id=run.id, symbol=symbol, company=cand["company"], signal_type="swing", direction=swing.direction, status="approved" if srd["approved"] else "rejected", risk_rating=srd["risk_rating"], price=quote["price"], confidence=swing_conf, probability=swing_prob, entry=srd["entry"], stop_loss=srd["stop_loss"], target1=srd["target1"], target2=srd["target2"], risk_reward=srd["risk_reward"], position_size=srd["position_size"], max_risk_usd=srd["max_risk_usd"], scores={"swing": swing.score, "technical": tech.score, "volatility": vol.score, "confidence": swing_conf}, rationale=[regime_note] + swing_notes + swing_risk.notes[:2], rejection_reasons=srd["rejection_reasons"], option_contract={}, indicators=tech.data.get("indicators", {}), history=[b["close"] for b in history[-30:]], )) created += 1 # ── Crypto engine (Platform Expansion v2 Module 4, feature-flagged) ── if os.getenv("CRYPTO_ENABLED", "false").lower() == "true": for item in provider.get_crypto_universe(): symbol = item["symbol"] history = provider.get_history(symbol, days=90) if not bars_are_fresh(history): stale_skipped += 1 logger.warning("CIRCUIT BREAKER: %s crypto data stale — skipped", symbol) continue quote = provider.get_quote(symbol) cctx = {"symbol": symbol, "history": history, "quote": quote} cswing = swing_agent.run(cctx) if not (cswing.direction and cswing.score >= MIN_SWING_SCORE): continue ctech = tech_agent.run(cctx) cnews = news_agent.run(cctx) confirms = (ctech.direction == "CALL") == (cswing.direction == "BUY") c_conf = cswing.score * 0.50 + ctech.score * (0.25 if confirms else 0.08) \ + cnews.score * 0.15 + vol_agent.run(cctx).score * 0.10 c_conf = round(max(0.0, min(98.0, c_conf)), 1) c_prob = round(max(5.0, min(90.0, 23 + c_conf * 0.5)), 1) if calibration and calibration.get("active"): from ..services.learning import adjust_probability c_prob = adjust_probability(c_prob, calibration) crisk = risk_agent.run({ **cctx, "direction": cswing.direction, "atr": cswing.data.get("atr"), "atr_mult": STOP_ATR_MULT, "confidence": c_conf, "probability": c_prob, "open_risk_usd": open_risk_usd, "risk_scale": risk_scale, }) crd = crisk.data if crd["approved"]: open_risk_usd += crd["max_risk_usd"] db.add(models.Signal( scan_run_id=run.id, symbol=symbol, company=item["company"], signal_type="crypto", direction=cswing.direction, status="approved" if crd["approved"] else "rejected", risk_rating=crd["risk_rating"], price=quote["price"], confidence=c_conf, probability=c_prob, entry=crd["entry"], stop_loss=crd["stop_loss"], target1=crd["target1"], target2=crd["target2"], risk_reward=crd["risk_reward"], position_size=crd["position_size"], max_risk_usd=crd["max_risk_usd"], scores={"swing": cswing.score, "technical": ctech.score, "news": cnews.score, "confidence": c_conf}, rationale=[regime_note, "24/7 crypto market — no options chain or earnings gate"] + cswing.notes + cnews.notes[:3] + crisk.notes[:2], rejection_reasons=crd["rejection_reasons"], option_contract={}, indicators=ctech.data.get("indicators", {}), history=[b["close"] for b in history[-30:]], )) created += 1 run.signals_created = created if stale_skipped: logger.warning("Scan #%s: circuit breaker skipped %s stale symbol(s)", run.id, stale_skipped) db.commit() db.refresh(run) return run def _current_open_risk(db: Session) -> float: """Sum of risk on currently open paper trades.""" open_trades = db.query(models.PaperTrade).filter(models.PaperTrade.status == "open").all() return sum(abs(t.entry_price - t.stop_loss) * t.quantity for t in open_trades)