""" ========================================= CV6 Autonomous AI Trading Operating System Main Engine — Orchestrates the full cycle: 1. SCANNING — scan full market 2. RANKING — score & rank candidates 3. ANALYZING — AI consensus per top pick 4. RISK CHECK — validate trade feasibility 5. EXECUTING — place orders via correct broker 6. MONITORING — manage SL, trail, exit 7. LEARNING — log results for improvement Repeats continuously until market close or STOP. ========================================= """ import asyncio import random import threading import time import uuid from datetime import datetime from typing import Any, Dict, List, Optional from loguru import logger from app.autonomous.autonomous_config import AutonomousConfig, load_config, save_config from app.autonomous.capital_pool import AutonomousPosition, CapitalPoolManager from app.autonomous.market_scanner import MarketScanner, ScanResult, ALL_NSE_SYMBOLS as _ALL_NSE from app.autonomous.broker_allocator import BrokerAllocator from app.autonomous.position_monitor import position_monitor from app.autonomous.ai_cache import ai_cache, CachedAIResult from app.autonomous.smart_scanner import SmartScanner, SmartCandidate from app.risk_guard.risk_guard import risk_guard from app.learning.learning_service import learning_service from app.fno.fno_engine import fno_engine from app.state.trading_state_manager import trading_state_manager from app.state.trading_events import TradingEvent, EventType # ── Engine state ───────────────────────────────────────────────────────────── class EngineState: IDLE = "IDLE" SCANNING = "SCANNING" RANKING = "RANKING" ANALYZING = "ANALYZING" EXECUTING = "EXECUTING" MONITORING = "MONITORING" PAUSED = "PAUSED" STOPPED = "STOPPED" ERROR = "ERROR" MAX_LOG = 200 # keep last 200 decision log entries class DecisionLog: """Rolling log of engine decisions and actions.""" def __init__(self): self._entries: List[Dict] = [] def add(self, event: str, detail: str, symbol: str = "", data: Any = None) -> None: entry = { "id": str(uuid.uuid4())[:8], "ts": time.strftime("%H:%M:%S"), "event": event, "symbol": symbol, "detail": detail, "data": data, } self._entries.insert(0, entry) if len(self._entries) > MAX_LOG: self._entries.pop() def recent(self, n: int = 50) -> List[Dict]: return self._entries[:n] def clear(self) -> None: self._entries.clear() class AutonomousEngine: """ Single background-thread engine. Uses asyncio inside the thread for AI calls. """ def __init__(self): self.config = load_config() self.state = EngineState.IDLE self.pool_manager = CapitalPoolManager() self.scanner = MarketScanner() self.smart_scanner = SmartScanner() self.allocator = BrokerAllocator(self.config.broker_allocation) self.log = DecisionLog() self._thread: Optional[threading.Thread] = None self._stop_event = threading.Event() self._lock = threading.Lock() self.started_at: Optional[float] = None self.cycle_count = 0 self.total_trades = 0 self.total_pnl = 0.0 self.scan_results: List[ScanResult] = [] self.smart_results: List[SmartCandidate] = [] self._loop: Optional[asyncio.AbstractEventLoop] = None # ── Duplicate order prevention ──────────────────────────────────────── # Symbols currently being executed (in-flight) — cleared after success/fail self._in_flight_symbols: set = set() # Symbols with recently placed orders: symbol → epoch timestamp # Prevents placing another order within ORDER_COOLDOWN_SECS of the last one self._recent_orders: dict = {} # symbol → float (epoch) self.ORDER_COOLDOWN_SECS = 900 # 15 minutes per symbol # ── Public control API ──────────────────────────────────────────────────── def start(self, config: Optional[AutonomousConfig] = None) -> dict: with self._lock: if self._thread and self._thread.is_alive(): return {"success": False, "message": "Engine already running"} if config: self.config = config save_config(config) self._stop_event.clear() self.log.clear() self.pool_manager.initialize( self.config.total_capital, self.config.capital_allocation, self.config.broker_allocation, ) # PHASE-2.5 FIX (Goal 4): pool_manager.initialize() above wipes # all in-memory positions — immediately reconcile against the # broker's actual live position book so a restart-then-start # doesn't leave CV6 blind to real positions opened before the # restart (which previously meant a genuine duplicate entry # could be placed, since symbol_has_open_position() only sees # what's in pool_manager). self.reconcile_broker_positions() self.allocator.update_map(self.config.broker_allocation) # Sync risk guard limits from config risk_guard.update_config({ "daily_loss_limit": self.config.max_daily_loss, "weekly_loss_limit": self.config.max_weekly_loss, "monthly_loss_limit": self.config.max_monthly_loss, "max_drawdown_pct": self.config.max_drawdown_pct, "circuit_filter_pct": 4.5 if self.config.circuit_filter else 100.0, "vix_block_threshold": self.config.vix_block_threshold, "news_filter_active": self.config.news_filter, }) # Sync AI cache TTL if self.config.ai_cache_ttl_seconds > 0: ai_cache.set_ttl(self.config.ai_cache_ttl_seconds) self.cycle_count = 0 self.total_trades = 0 self.started_at = time.time() self.state = EngineState.SCANNING scanner_type = "SmartScanner" if self.config.use_smart_scanner else "BasicScanner" self.log.add("START", f"Engine started — mode={self.config.mode}, capital=₹{self.config.total_capital:,.0f}, scanner={scanner_type}") self._thread = threading.Thread(target=self._run_loop, daemon=True, name="CV6-AutonomousEngine") self._thread.start() return {"success": True, "message": f"Autonomous engine started in {self.config.mode} mode"} def stop(self) -> dict: self._stop_event.set() self.state = EngineState.STOPPED self.log.add("STOP", "Engine stopped by user") return {"success": True, "message": "Engine stopping after current cycle"} def pause(self) -> dict: self.state = EngineState.PAUSED self.log.add("PAUSE", "Engine paused") return {"success": True, "message": "Engine paused"} def resume(self) -> dict: if self.state == EngineState.PAUSED: self.state = EngineState.SCANNING self.log.add("RESUME", "Engine resumed") return {"success": True, "message": "Engine resumed"} def status(self) -> dict: running = self._thread is not None and self._thread.is_alive() uptime = round(time.time() - self.started_at, 0) if self.started_at else 0 return { "state": self.state, "running": running, "mode": self.config.mode, "cycle": self.cycle_count, "total_trades": self.total_trades, "total_pnl": round(self.pool_manager.total_unrealized_pnl() + self.pool_manager.total_realized_pnl(), 2), "open_positions": len(self.pool_manager.all_positions()), "uptime_s": uptime, "scan_interval":self.config.scan_interval_seconds, "capital": self.config.total_capital, "pools": self.pool_manager.summary(), "last_scan_symbols": len(self.scan_results), "started_at": self.started_at, } def update_config(self, new_config: AutonomousConfig) -> dict: self.config = new_config save_config(new_config) self.allocator.update_map(new_config.broker_allocation) # Re-initialize capital pools immediately so new total_capital takes effect # without requiring a full engine restart self.pool_manager.initialize( new_config.total_capital, new_config.capital_allocation, new_config.broker_allocation, ) # PHASE-2.5 FIX (Goal 4): this also wipes pools — reconcile again. self.reconcile_broker_positions() logger.info( f"[Engine] Capital pools re-initialized: ₹{new_config.total_capital:,.0f} " f"({new_config.capital_allocation})" ) return {"success": True, "message": "Configuration updated — capital pools re-initialized"} # ── PHASE-2.5 FIX (Goal 4): startup/restart position reconciliation ─────── def reconcile_broker_positions(self) -> dict: """ Query every connected broker's real position book and restore any open position not already tracked in pool_manager. Without this, pool_manager.initialize() (called on every start()/update_config()) wipes local memory of positions that still genuinely exist at the broker — leaving Portfolio/Positions/Capital Pool blind to real exposure after any restart, and allowing symbol_has_open_position() to wrongly permit a duplicate entry in an already-open symbol. Best-effort by necessity: the original style (INTRADAY/SWING/FNO/ SCALPING/LONG_TERM), stop-loss and target of a pre-existing position cannot be recovered after a restart, since CV6 never persisted them anywhere. Style is inferred from the broker's own producttype; stop-loss/target are (re)estimated via the same style-based defaults used for new entries, so a reconciled position is never left with NO protective levels — but they may not match whatever was originally intended for that trade. """ restored, skipped, errors = 0, 0, [] try: from app.api.broker_api import manager as broker_mgr except Exception as e: return {"success": False, "message": f"broker manager unavailable: {e}"} already_tracked = {p.symbol for p in self.pool_manager.all_positions()} # Every distinct connected broker instance known to BrokerManager seen_ids = set() broker_instances = [] if broker_mgr.connected and broker_mgr.broker is not None: broker_instances.append((broker_mgr.broker_name or "unknown", broker_mgr.broker)) seen_ids.add(id(broker_mgr.broker)) for name, inst in getattr(broker_mgr, "_extra_brokers", {}).items(): if inst is not None and id(inst) not in seen_ids and getattr(inst, "is_connected", lambda: False)(): broker_instances.append((name, inst)) seen_ids.add(id(inst)) for broker_name, broker in broker_instances: try: raw = broker.positions() except Exception as e: errors.append(f"{broker_name}: {e}") continue if isinstance(raw, dict): if raw.get("success") is False: continue pos_list = raw.get("data") if isinstance(raw.get("data"), list) else [] elif isinstance(raw, list): pos_list = raw else: pos_list = [] for p in pos_list: symbol = str(p.get("tradingsymbol") or p.get("symbol", "")).upper() if not symbol: continue qty = p.get("netqty") if qty is None: qty = p.get("quantity", 0) try: qty = int(float(qty)) except (TypeError, ValueError): qty = 0 if qty == 0: continue if symbol in already_tracked: skipped += 1 continue avg_price = p.get("averageprice") or p.get("average_price") or 0 ltp = p.get("ltp") or p.get("lastprice") or avg_price try: avg_price = float(avg_price) ltp = float(ltp) except (TypeError, ValueError): avg_price, ltp = 0.0, 0.0 if avg_price <= 0: continue exchange = str(p.get("exchange", "NSE")).upper() side = "BUY" if qty > 0 else "SELL" product = str(p.get("producttype") or p.get("product", "")).upper() if product in ("INTRADAY", "MIS"): style = "INTRADAY" elif "FNO" in exchange or exchange in ("NFO", "MCX", "BFO", "CDS"): style = "FNO" else: # DELIVERY/CARRYFORWARD/CNC/unknown — safest default so # this doesn't get force-squared-off at 3:10pm like an # INTRADAY position would if it was really meant to carry. style = "SWING" pool = self.pool_manager.get_pool(style) if pool is None: skipped += 1 continue sl, tp = self.allocator.calculate_targets(ltp or avg_price, side, style) pos = AutonomousPosition( symbol=symbol, style=style, side=side, qty=abs(qty), entry_price=avg_price, stop_loss=sl, target_price=tp, broker=broker_name, order_id="", exchange=exchange, ) pos.strategy = "RECONCILED" pool.add_position(pos) already_tracked.add(symbol) restored += 1 self.log.add( "RECONCILED", f"Restored real {side} {abs(qty)} {symbol} @ ₹{avg_price:.2f} " f"from {broker_name} (style={style}, SL/TP estimated — " f"original levels unknown after restart)", symbol=symbol, ) if pool.available_capital < 0: logger.warning( f"[Engine] Pool '{style}' available_capital went negative " f"({pool.available_capital:.0f}) after reconciling {symbol} — " f"real broker exposure exceeds this pool's configured allocation." ) if restored or errors: logger.info(f"[Engine] Position reconciliation: restored={restored} skipped={skipped} errors={errors}") return {"success": True, "restored": restored, "skipped": skipped, "errors": errors} def bootstrap_pools_and_reconcile(self) -> dict: """ PHASE-2.5 FIX (Goal 4): "on application startup, reconcile broker positions with local state" — this must work even before the user manually clicks Start. Capital pools normally only exist once start()/update_config() has run; this initializes them from the last-saved config (without touching engine run-state — the scanning/trading loop is NOT started here) so Portfolio, Positions and Capital Pool are accurate immediately after a server restart, reflecting real broker exposure even while the engine sits idle. Safe/cheap to call multiple times. """ self.pool_manager.initialize( self.config.total_capital, self.config.capital_allocation, self.config.broker_allocation, ) return self.reconcile_broker_positions() # ── Background run loop ─────────────────────────────────────────────────── def _run_loop(self) -> None: self._loop = asyncio.new_event_loop() asyncio.set_event_loop(self._loop) try: self._loop.run_until_complete(self._async_loop()) except Exception as e: logger.error(f"[AutonomousEngine] Fatal error in run loop: {e}") self.state = EngineState.ERROR self.log.add("ERROR", str(e)) finally: self._loop.close() self.state = EngineState.STOPPED async def _async_loop(self) -> None: while not self._stop_event.is_set(): if self.state == EngineState.PAUSED: await asyncio.sleep(1) continue if not self._is_trading_hours(): if self.state not in (EngineState.STOPPED, EngineState.PAUSED): self.state = EngineState.PAUSED self.log.add("MARKET_CLOSED", "Outside trading hours — engine paused") await asyncio.sleep(30) continue cycle_start = time.time() self.cycle_count += 1 try: # ── Phase 1: SCAN ──────────────────────────────────────── self.state = EngineState.SCANNING await self._phase_scan() # ── Phase 2: RANK ──────────────────────────────────────── self.state = EngineState.RANKING candidates = self._phase_rank() # ── Phase 3: AI ANALYSIS ───────────────────────────────── if candidates and not self._stop_event.is_set(): self.state = EngineState.ANALYZING await self._phase_analyze(candidates[:5]) # ── Phase 4: EXECUTE ───────────────────────────────────── if not self._stop_event.is_set(): self.state = EngineState.EXECUTING await self._phase_execute(candidates[:3]) # ── Phase 5: MONITOR ───────────────────────────────────── if not self._stop_event.is_set(): self.state = EngineState.MONITORING self._phase_monitor() except Exception as e: logger.error(f"[Engine cycle {self.cycle_count}] Error: {e}") self.log.add("CYCLE_ERROR", str(e)) # Wait until next scan interval elapsed = time.time() - cycle_start wait = max(0, self.config.scan_interval_seconds - elapsed) if wait > 0 and not self._stop_event.is_set(): await asyncio.sleep(wait) self.state = EngineState.STOPPED # ── Phases ──────────────────────────────────────────────────────────────── async def _phase_scan(self) -> None: """Scan full market using SmartScanner (5-stage funnel) or basic scanner.""" loop = asyncio.get_event_loop() if self.config.use_smart_scanner: self.log.add("SCAN", f"SmartScanner: Stage1({len(_ALL_NSE)}) → Stage2(200) → Stage3(20) → Stage4(5) → Stage5(AI)…") smart_results = await loop.run_in_executor( None, lambda: self.smart_scanner.scan( strategies_enabled=self.config.strategies_enabled, min_score=self.config.min_score_to_trade, max_sector_pct=self.config.max_sector_concentration, circuit_filter=self.config.circuit_filter, volatility_filter=self.config.volatility_filter, ) ) self.smart_results = smart_results # Convert SmartCandidate → ScanResult for downstream phases converted = [] for c in smart_results: sr = ScanResult(c.symbol, { "ltp": c.ltp, "open": c.ltp, "high": c.ltp * 1.005, "low": c.ltp * 0.995, "change_pct": c.change_pct, "volume": c.volume, "source": "smart_scanner", }) sr.opportunity_score = c.total_score sr.signal = c.signal sr.strategy = c.strategy converted.append(sr) self.scan_results = converted self.log.add("SCAN_DONE", f"SmartScanner → {len(smart_results)} top candidates") else: self.log.add("SCAN", f"Scanning {len(_ALL_NSE)} symbols (basic scanner)…") results = await loop.run_in_executor( None, lambda: self.scanner.scan( strategies_enabled=self.config.strategies_enabled, min_score=self.config.min_score_to_trade, ) ) self.scan_results = results self.log.add("SCAN_DONE", f"Found {len(results)} tradeable candidates (score≥{self.config.min_score_to_trade})") def _phase_rank(self) -> List[ScanResult]: """Rank and filter top candidates.""" candidates = self.scanner.top_candidates(n=10, min_score=self.config.min_score_to_trade) # Filter out already-traded symbols traded_syms = {p.symbol for p in self.pool_manager.all_positions()} candidates = [c for c in candidates if c.symbol not in traded_syms] # Respect max daily trades total_daily = sum(pool.daily_trades for pool in self.pool_manager.pools.values()) if total_daily >= self.config.max_trades_per_day: self.log.add("LIMIT", f"Max daily trades reached ({self.config.max_trades_per_day})") return [] if candidates: self.log.add("RANK", f"Top: {candidates[0].symbol} score={candidates[0].opportunity_score:.1f} signal={candidates[0].signal}") return candidates async def _phase_analyze(self, candidates: List[ScanResult]) -> None: """Run AI consensus on top candidates. Checks cache first to avoid duplicate calls.""" if not candidates: return for candidate in candidates: # ── Cache check ───────────────────────────────────────────────── cached = ai_cache.get(candidate.symbol) if cached and self.config.ai_cache_ttl_seconds > 0: candidate._ai_signal = cached.signal candidate._ai_confidence = cached.confidence candidate._ai_reason = f"[CACHE] {cached.reason}" self.log.add("AI_CACHED", f"Cache hit → {cached.signal} ({cached.confidence:.0f}%)", symbol=candidate.symbol) continue # ── TIER-1 (1B): REAL technicals before the AI decides ─────────── # Compute RSI/EMA/VWAP/ATR/Supertrend from REAL OHLCV candles and # overwrite the candidate's (previously random-fabricated) values so # the AI sees real data. In REAL mode, if real data can't be fetched # we refuse to trade on synthetic data — skip the candidate entirely # (this also saves the AI call). candidate._real_data_ok = False try: from app.market.technicals import compute_technicals rt = compute_technicals(candidate.symbol, getattr(candidate, "exchange", "NSE")) candidate._real_technicals = rt if rt.is_real: candidate._real_data_ok = True candidate.rsi = rt.rsi candidate.ema_9 = rt.ema_9 candidate.ema_21 = rt.ema_21 candidate.vwap = rt.vwap candidate.volume_ratio = rt.volume_ratio candidate.atr_pct = rt.atr_pct candidate.supertrend = rt.supertrend except Exception as _te: candidate._real_technicals = None self.log.add("TECHNICALS_ERR", f"Real technicals failed: {str(_te)[:60]}", symbol=candidate.symbol) if self.config.mode == "REAL" and not candidate._real_data_ok: candidate._ai_signal = "HOLD" self.log.add( "SKIP_SYNTHETIC_DATA", "Real indicators unavailable — refusing to trade on fabricated data", symbol=candidate.symbol, ) continue try: ai_result = await self._run_ai_consensus(candidate) signal = ai_result.get("consensus", "HOLD") confidence = float(ai_result.get("confidence", 0)) reason = ai_result.get("reason", "") candidate._ai_signal = signal candidate._ai_confidence = confidence candidate._ai_reason = reason # Store in cache if self.config.ai_cache_ttl_seconds > 0: ai_cache.put(candidate.symbol, CachedAIResult( symbol=candidate.symbol, signal=signal, confidence=confidence, reason=reason, models_used=self.config.enabled_ai_models, )) self.log.add( "AI_SIGNAL", f"AI → {signal} ({confidence:.0f}% conf): {reason[:60]}", symbol=candidate.symbol, ) except Exception as e: candidate._ai_signal = candidate.signal candidate._ai_confidence = candidate.opportunity_score candidate._ai_reason = "AI unavailable — using technical signal" self.log.add("AI_FALLBACK", f"AI unavailable for {candidate.symbol}: {str(e)[:60]}", symbol=candidate.symbol) async def _run_ai_consensus(self, candidate: ScanResult) -> dict: """Call AI consensus engine (async) with rich context package.""" try: from app.ai.ai_consensus import consensus_engine # ── Phase 4 Enhancement: Build rich AI context package ──────────── # Attach rank_score from smart_results if available (set by SmartRanker) smart_cand = None if hasattr(self, "smart_results"): for sc in self.smart_results: if sc.symbol == candidate.symbol: smart_cand = sc break context_package: Optional[dict] = None try: from app.autonomous.ai_context_builder import ai_context_builder ctx = ai_context_builder.build( candidate = smart_cand if smart_cand is not None else candidate, portfolio_state = self.pool_manager, ) context_package = {"prompt_text": ctx.to_prompt()} self.log.add( "AI_CONTEXT", f"Context built: market={ctx.market.personality} " f"MTF={ctx.mtf.agreement_pct:.0f}% " f"confidence={ctx.confidence.overall_confidence:.0f}", symbol=candidate.symbol, ) except Exception as ctx_err: self.log.add("AI_CTX_WARN", f"Context builder skipped: {str(ctx_err)[:60]}", symbol=candidate.symbol) result = await consensus_engine.consensus( symbol=candidate.symbol, market_data={ "ltp": candidate.ltp, "open": candidate.open, "high": candidate.high, "low": candidate.low, "change_pct": candidate.change_pct, "volume": candidate.volume, }, context_package=context_package, ) d = result.to_dict() return { "consensus": d.get("signal", "HOLD"), "confidence": d.get("confidence", 0), "reason": d.get("reason", ""), } except Exception as e: # If no API key or error — fall back to technical signal return { "consensus": candidate.signal, "confidence": candidate.opportunity_score, "reason": f"Technical signal (AI unavailable: {str(e)[:40]})", } async def _phase_execute(self, candidates: List[ScanResult]) -> None: """Execute orders for approved candidates.""" total_open = len(self.pool_manager.all_positions()) if total_open >= self.config.max_open_positions: self.log.add("LIMIT", f"Max open positions ({self.config.max_open_positions}) reached — skipping execution") return # ── Nifty Macro Filter ──────────────────────────────────────────────── # Check overall market direction via Nifty 50 structure. # If Nifty is in DOWNTREND + below 200 EMA, block all BUY signals. # If Nifty is in UPTREND + above 200 EMA, block all SELL signals. _nifty_blocks_buy = False _nifty_blocks_sell = False try: from app.market.market_structure import get_nifty_structure _nifty_ms = get_nifty_structure() if _nifty_ms: if _nifty_ms.trend == "DOWNTREND" and not _nifty_ms.above_200ema: _nifty_blocks_buy = True self.log.add( "MACRO_FILTER", f"Nifty DOWNTREND below 200 EMA ₹{_nifty_ms.ema_200:.0f} — all BUY signals blocked" ) elif _nifty_ms.trend == "UPTREND" and _nifty_ms.above_200ema: _nifty_blocks_sell = True self.log.add( "MACRO_FILTER", f"Nifty UPTREND above 200 EMA ₹{_nifty_ms.ema_200:.0f} — all SELL signals blocked" ) except Exception as _nifty_err: logger.debug(f"[Engine] Nifty macro filter skipped: {_nifty_err}") # Purge stale cooldown entries now_ts = time.time() self._recent_orders = { sym: ts for sym, ts in self._recent_orders.items() if now_ts - ts < self.ORDER_COOLDOWN_SECS } for candidate in candidates: if self._stop_event.is_set(): break # ── Duplicate order / duplicate position guard ──────────────────── sym = candidate.symbol if sym in self._in_flight_symbols: self.log.add("SKIP_INFLIGHT", "Order already in-flight for this symbol", symbol=sym) continue if sym in self._recent_orders: age = int(now_ts - self._recent_orders[sym]) self.log.add("SKIP_COOLDOWN", f"Order placed {age}s ago — cooldown {self.ORDER_COOLDOWN_SECS}s", symbol=sym) continue # PHASE-2 FIX: true "already have an open position" check — the # in-flight/cooldown guards above only prevent re-ordering within a # short window; this blocks a second position in the same symbol # for as long as the first one stays open, across any pool/style. # The TSM check is the AUTHORITATIVE one (survives restart, covers # manual + reconciled positions the pool may not know about yet); # the pool check is a fast local backstop. if self.config.mode == "REAL" and trading_state_manager.has_open_position(sym): self.log.add("SKIP_DUPLICATE_POSITION", "Symbol already has an open position (source of truth)", symbol=sym) continue if self.pool_manager.symbol_has_open_position(sym): self.log.add("SKIP_DUPLICATE_POSITION", "Symbol already has an open autonomous position", symbol=sym) continue # ── TIER-1 (1B) defense-in-depth: never place a REAL trade on a # candidate whose indicators weren't verified as real in the # analyze phase. _phase_analyze already skips these, but this is a # hard backstop at the execution boundary. if self.config.mode == "REAL" and not getattr(candidate, "_real_data_ok", False): self.log.add("SKIP_SYNTHETIC_DATA", "Execution gate: indicators not verified real — not trading", symbol=sym) continue ai_signal = candidate._ai_signal or candidate.signal ai_confidence = candidate._ai_confidence or candidate.opportunity_score ai_reason = candidate._ai_reason # ── Nifty Macro Override ────────────────────────────────────────── if ai_signal == "BUY" and _nifty_blocks_buy: self.log.add( "SKIP_MACRO", "BUY blocked — Nifty macro DOWNTREND", symbol=candidate.symbol, ) continue if ai_signal == "SELL" and _nifty_blocks_sell: self.log.add( "SKIP_MACRO", "SELL blocked — Nifty macro UPTREND", symbol=candidate.symbol, ) continue # ── 200 EMA Structure Filter ────────────────────────────────────── # Only BUY when price is above 200 EMA. # Only SELL when price is below 200 EMA. # If market structure says WAIT, skip regardless of AI signal. try: from app.market.market_structure import analyse_structure _ms = analyse_structure(candidate.symbol) if _ms: # Trade bias override: WAIT means skip if _ms.trade_bias == "WAIT": self.log.add( "SKIP_STRUCTURE", f"Structure bias=WAIT (price near S/R) — skipping", symbol=candidate.symbol, ) continue # 200 EMA direction filter if ai_signal == "BUY" and not _ms.above_200ema: self.log.add( "SKIP_200EMA", f"BUY rejected — price ₹{_ms.current_price:.2f} BELOW 200 EMA ₹{_ms.ema_200:.2f}", symbol=candidate.symbol, ) continue if ai_signal == "SELL" and _ms.above_200ema: self.log.add( "SKIP_200EMA", f"SELL rejected — price ₹{_ms.current_price:.2f} ABOVE 200 EMA ₹{_ms.ema_200:.2f}", symbol=candidate.symbol, ) continue except Exception as _ema_err: logger.debug(f"[Engine] 200 EMA filter skipped for {candidate.symbol}: {_ema_err}") # ── Volume Confirmation Filter ──────────────────────────────────── # Reject trades when volume signal is DIVERGENCE # (price moved but low volume = fake/weak breakout) try: from app.market.volume_analyzer import analyse_volume _va = analyse_volume(candidate.symbol) if _va and _va.price_volume == "DIVERGENCE": self.log.add( "SKIP_VOL_DIVERGENCE", f"Volume DIVERGENCE — price moved but vol only {_va.volume_ratio:.1f}x avg (fake/weak move)", symbol=candidate.symbol, ) continue except Exception as _vol_err: logger.debug(f"[Engine] Volume filter skipped for {candidate.symbol}: {_vol_err}") # Skip if AI below threshold if ai_confidence < self.config.consensus_threshold: self.log.add( "SKIP_LOW_CONF", f"AI confidence {ai_confidence:.0f}% < threshold {self.config.consensus_threshold:.0f}%", symbol=candidate.symbol, ) continue # Skip HOLD/NEUTRAL signals if ai_signal in ("HOLD", "NEUTRAL"): continue # Determine style based on strategy style = self._map_style(candidate.strategy) pool = self.pool_manager.get_pool(style) if pool is None: continue broker = self._resolve_connected_broker(style) exchange = self.allocator.get_exchange(style) product = self.allocator.get_product_type(style) side = ai_signal if ai_signal in ("BUY", "SELL") else ("BUY" if candidate.change_pct > 0 else "SELL") # ── F&O mode: use FnoEngine for option strike + lot size ──────── if style == "FNO" and self.config.fno_enabled: try: option_type = self.config.fno_option_type # CE | PE | BOTH # Map BUY→CE, SELL→PE when BOTH is selected if option_type == "BOTH": option_type = "CE" if side == "BUY" else "PE" # Step 1: Get option chain for this symbol chain = fno_engine.get_option_chain(candidate.symbol, "current", candidate.ltp) # Step 2: Select best strike (ATM/ITM/OTM per config) opt_contract = fno_engine.select_strike( chain = chain, option_type = option_type, moneyness = self.config.fno_moneyness, ) lot_size = fno_engine.get_lot_size(candidate.symbol) # Use option premium as trade price trade_price = opt_contract.premium if opt_contract else candidate.ltp sl = round(trade_price * 0.80, 2) # 20% SL on premium tp = round(trade_price * 1.50, 2) # 50% target on premium qty = lot_size # 1 lot minimum for F&O val = trade_price * qty exchange = "NFO" strike_str = f"{opt_contract.strike} {option_type}" if opt_contract else "N/A" self.log.add( "FNO_STRIKE", f"Strike {strike_str} | {self.config.fno_moneyness} | lot={lot_size} | premium=₹{trade_price:.2f}", symbol=candidate.symbol, ) except Exception as fno_err: self.log.add("FNO_ERR", f"FnoEngine error: {str(fno_err)[:60]} — falling back to equity mode", symbol=candidate.symbol) sl, tp = self.allocator.calculate_targets(candidate.ltp, side, style) qty, val = self.allocator.calculate_position_size( pool.available_capital, candidate.ltp, self.config.risk_pct, sl, style ) else: sl, tp = self.allocator.calculate_targets(candidate.ltp, side, style) qty, val = self.allocator.calculate_position_size( pool.available_capital, candidate.ltp, self.config.risk_pct, sl, style ) ok, reason = pool.can_trade(val, self.config.max_daily_loss, self.config.max_open_positions) if not ok: self.log.add("BLOCKED", reason, symbol=candidate.symbol) continue # ── Capital authority: the TSM (source of truth) has the final say # on available capital for REAL trades, so the go/no-go decision can # never exceed real, persisted, cross-path capital — even if the # engine's internal pool has drifted from it during the session. if self.config.mode == "REAL": _tsm_avail = trading_state_manager.snapshot()["capital"]["available"] if val > _tsm_avail: self.log.add( "BLOCKED", f"Insufficient real capital (need ₹{val:.0f}, " f"source-of-truth available ₹{_tsm_avail:.0f})", symbol=candidate.symbol, ) continue # ── Risk Guard check ───────────────────────────────────────────── open_pos_dicts = [ {"sector": getattr(p, "sector", "Other")} for p in self.pool_manager.all_positions() ] verdict = risk_guard.check_entry( symbol = candidate.symbol, sector = getattr(candidate, "sector", "Other"), ltp = candidate.ltp, change_pct = candidate.change_pct, vix = 0.0, # real VIX from market feed if available news_sentiment = "NEUTRAL", # real sentiment from news filter if available open_positions = open_pos_dicts, capital = self.config.total_capital, deployed = val, mode = self.config.mode, exchange = exchange, # NSE / MCX — drives market-hours check ) if not verdict.allowed: self.log.add("RISK_BLOCKED", f"[{verdict.layer}] {verdict.reason}", symbol=candidate.symbol) continue # ── Place order ─────────────────────────────────────────────── self._in_flight_symbols.add(sym) try: fill_status, order_id, order_msg, filled_qty, avg_price, sl_order_id = await self._place_order( broker=broker, symbol=candidate.symbol, exchange=exchange, side=side, qty=qty, price=candidate.ltp, product=product, stop_loss=sl, target_price=tp, ) finally: # Always release in-flight lock; record cooldown on success or failure self._in_flight_symbols.discard(sym) self._recent_orders[sym] = time.time() # PHASE-2.5 FIX (Goal 2): only create/track a local position once # the broker has CONFIRMED a fill (FILLED or PARTIALLY_FILLED) — # "accepted" alone is not enough, and a rejected/cancelled/still- # pending order must never reserve capital or open a position. if fill_status not in ("FILLED", "PARTIALLY_FILLED"): self.log.add( "ORDER_REJECTED", f"{side} {qty} {candidate.symbol} not filled ({fill_status}) by {broker}: {order_msg[:80]}", symbol=candidate.symbol, data={"broker": broker, "qty": qty, "price": candidate.ltp, "fill_status": fill_status}, ) continue # PHASE-2.5 FIX (Goal 2): use the CONFIRMED filled quantity/price, # not the originally requested qty/candidate.ltp — a partial fill # must be recorded at its true size, not the intended size. actual_qty = filled_qty if filled_qty > 0 else qty actual_price = avg_price if avg_price > 0 else candidate.ltp if fill_status == "PARTIALLY_FILLED": self.log.add( "PARTIAL_FILL", f"{side} {candidate.symbol}: requested {qty}, filled {actual_qty} @ ₹{actual_price:.2f}", symbol=candidate.symbol, data={"requested_qty": qty, "filled_qty": actual_qty}, ) pos = AutonomousPosition( symbol=candidate.symbol, style=style, side=side, qty=actual_qty, entry_price=actual_price, stop_loss=sl, target_price=tp, broker=broker, order_id=order_id, exchange=exchange, # drives EOD close time (NSE vs MCX) ) pos.ai_confidence = ai_confidence pos.strategy = candidate.strategy pos.sl_order_id = sl_order_id # TIER-1: broker-side SL order id if self.config.mode == "REAL" and not sl_order_id: self.log.add( "SL_UNPROTECTED", f"Entry filled but broker STOP-LOSS not placed — software monitor is sole protection", symbol=candidate.symbol, ) pool.add_position(pos) self.total_trades += 1 # ── SOURCE OF TRUTH: emit ONE fill event to the TradingStateManager. # TSM is the authoritative, persisted owner of REAL orders/positions/ # capital/PnL that every read path (dashboard, /state API, journal, # analytics, websocket) consumes. It is REAL-ONLY — PAPER trades stay # in the paper engine sandbox so the two never mix in one ledger. # The capital_pool remains the engine's internal monitor scratch space. if self.config.mode == "REAL": try: trading_state_manager.apply_event(TradingEvent( event_type=EventType.ORDER_FILLED, symbol=candidate.symbol, broker=broker, exchange=exchange, side=side, order_id=order_id, quantity=qty, filled_quantity=actual_qty, avg_price=actual_price, fill_status=fill_status, style=style, strategy=candidate.strategy, stop_loss=sl, target_price=tp, sl_order_id=sl_order_id, ai_confidence=ai_confidence, mode="REAL", source="autonomous_engine", )) except Exception as _tsm_err: logger.error(f"[Engine] TSM fill event failed for {candidate.symbol}: {_tsm_err}") self.log.add( "TRADE", f"{side} {actual_qty} @ ₹{actual_price:.2f} | SL={sl:.2f} TP={tp:.2f} | {broker} | {ai_reason[:50]}", symbol=candidate.symbol, data={"order_id": order_id, "qty": actual_qty, "price": actual_price, "sl": sl, "tp": tp}, ) def _resolve_connected_broker(self, style: str) -> str: """ Return the allocator's preferred broker for this style, but verify it is connected. If it isn't, mark it failed and fall back through the allocator's priority list until a connected broker is found. Prevents silent order failures when AliceBlue/Upstox/Zerodha are not connected. """ from app.api.broker_api import manager as broker_mgr # PHASE-2 FIX (Goals 5 & 6): only real, implemented brokers belong in # the failover list — AliceBlue/Upstox/Zerodha/Groww never send # anything to any exchange (see app/broker/*.py), so failing over to # one of them previously looked like a working broker. _FALLBACK = ["AngelOne", "MStock"] def _is_connected(name: str) -> bool: try: inst = broker_mgr.get_broker(name) return bool(inst and getattr(inst, "is_connected", lambda: False)()) except Exception: return False # First try the allocator's chosen broker chosen = self.allocator.get_broker(style) if _is_connected(chosen): return chosen # Mark it failed so the allocator updates its internal state self.allocator.mark_failed(chosen) self.log.add( "BROKER_FALLBACK", f"Primary broker {chosen!r} not connected — trying fallback", ) # Walk the fallback list and return first connected broker for candidate in _FALLBACK: if candidate == chosen: continue if _is_connected(candidate): logger.info("[Engine] Broker fallback: %s → %s (style=%s)", chosen, candidate, style) return candidate # Nothing is connected — return the primary name so the order fails gracefully logger.warning("[Engine] No connected broker found; using %s (will fail gracefully)", chosen) return chosen async def _place_order( self, broker: str, symbol: str, exchange: str, side: str, qty: int, price: float, product: str, stop_loss: float = 0.0, target_price: float = 0.0, ) -> tuple[str, str, str, int, float, str]: """ Place order via broker or paper engine. Returns (fill_status, order_id, message, filled_qty, avg_price, sl_order_id). PHASE-2.5 FIX (Goal 2): fill_status reflects the broker's CONFIRMED outcome (FILLED/PARTIALLY_FILLED/REJECTED/CANCELLED/PENDING/UNKNOWN) — "accepted" is never treated as "filled". The real-mode path blocks (in a background-thread executor, not the FastAPI event loop) until ExecutionEngine/OrderManager's own poll_until_final resolves. TIER-1: passing stop_loss makes OrderManager place a broker-native STOPLOSS_MARKET order after the fill; sl_order_id is returned. """ # PHASE-2 FIX (Goal 3): idempotency key derived from the intended # order's identity + a coarse time bucket, so a re-evaluated signal or # retry within the same bucket can't place a second real order. idem_key = f"{broker}:{symbol}:{exchange}:{side}:{qty}:{int(time.time() // self.ORDER_COOLDOWN_SECS)}" try: if self.config.mode == "REAL": from app.execution.execution_engine import ExecutionEngine from app.execution.execution_models import ExecutionRequest, OrderSide, OrderType, ProductType eng = ExecutionEngine() req = ExecutionRequest( broker=broker, symbol=symbol, exchange=exchange, side=OrderSide(side), quantity=qty, order_type=OrderType.MARKET, product=ProductType.INTRADAY if product == "INTRADAY" else ProductType.DELIVERY, idempotency_key=idem_key, stop_loss=stop_loss or None, target_price=target_price or None, ) # OrderManager.place_order() blocks (fill-status polling via # time.sleep) — run it off this engine's own event loop so a # multi-second poll doesn't stall the scan/monitor cycle. loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, eng.execute, req) return (result.fill_status, result.order_id or "", result.message, result.filled_quantity, result.avg_fill_price, result.sl_order_id) else: # PAPER mode — use paper engine (always a full, instant fill) from app.paper_trading.paper_engine import paper_engine result = paper_engine.place_order( symbol=symbol, side=side, qty=qty, price=price, order_type="MARKET", ) success = bool(result.get("success", False)) order_id = result.get("order_id", "") if success else "" return ( "FILLED" if success else "REJECTED", order_id, result.get("message", ""), qty if success else 0, price if success else 0.0, "", ) except Exception as e: logger.warning(f"[Engine] Order placement failed ({broker}/{symbol}): {e}") self.allocator.mark_failed(broker) return "REJECTED", "", str(e), 0, 0.0, "" def _fetch_broker_positions_normalized(self) -> tuple[bool, list]: """Pull + normalize the live position book from all connected brokers. Returns (fetch_ok, positions). fetch_ok is False if NO broker could be queried — callers must not treat an empty list as 'broker is flat' unless fetch_ok is True.""" from app.api.broker_api import manager as _bm seen, out, any_ok = set(), [], False candidates = [] if _bm.connected and _bm.broker is not None: candidates.append((_bm.broker_name or "", _bm.broker)) for _n, _inst in getattr(_bm, "_extra_brokers", {}).items(): if _inst is not None and id(_inst) not in seen: candidates.append((_n, _inst)); seen.add(id(_inst)) for bname, binst in candidates: try: raw = binst.positions() except Exception: continue any_ok = True plist = raw.get("data") if isinstance(raw, dict) else (raw if isinstance(raw, list) else []) for p in (plist or []): q = p.get("netqty", p.get("quantity", 0)) try: q = int(float(q)) except (TypeError, ValueError): q = 0 out.append({ "symbol": str(p.get("tradingsymbol") or p.get("symbol", "")), "exchange": str(p.get("exchange", "NSE")), "broker": bname, "netqty": q, "avg_price": float(p.get("averageprice") or p.get("average_price") or 0), "ltp": float(p.get("ltp") or p.get("lastprice") or 0), }) return any_ok, out _RECONCILE_EVERY_CYCLES = 20 # ~10 min at the 30s default scan interval def _phase_monitor(self) -> None: """Update positions, trail SL, check exits.""" # ── Intra-session broker reconciliation (REAL only) ────────────────── # Periodically realign the source of truth with the broker's real book: # restores anything opened out-of-band and force-closes positions the # broker has exited behind our back (manual close, exchange SL/square-off). if self.config.mode == "REAL" and self.cycle_count % self._RECONCILE_EVERY_CYCLES == 0: # TIER-2 (2A): proactively validate/refresh each broker session # BEFORE reconciling, so a mid-day token expiry self-heals instead # of silently failing every subsequent call. try: from app.api.broker_api import manager as _bm for _inst in ([_bm.broker] + list(getattr(_bm, "_extra_brokers", {}).values())): if _inst is not None and hasattr(_inst, "validate_session"): _inst.validate_session() except Exception as _vs_err: logger.warning(f"[Engine] session validation failed: {_vs_err}") try: fetch_ok, bpos = self._fetch_broker_positions_normalized() if fetch_ok: trading_state_manager.reconcile_with_broker(bpos, close_missing=True) except Exception as _rec_err: logger.warning(f"[Engine] periodic reconcile failed: {_rec_err}") if not self.pool_manager.all_positions(): return # Build price map from last scan price_map = {r.symbol: r.ltp for r in self.scan_results} # ── SOURCE OF TRUTH: feed live prices into TSM so its unrealized PnL # (shown on dashboard / /state / websocket) stays current. for _sym, _ltp in price_map.items(): if _ltp and trading_state_manager.has_open_position(_sym): try: trading_state_manager.apply_event(TradingEvent( event_type=EventType.PRICE_TICK, symbol=_sym, ltp=float(_ltp), source="autonomous_engine", )) except Exception: pass for pool in self.pool_manager.pools.values(): actions = position_monitor.monitor_pool( pool, price_map, self.config.trading_end, getattr(self.config, "commodity_trading_end", "23:00"), ) for action in actions: if action["action"] == "CLOSE": pnl = action["pnl"] reason = action["reason"] sym = action.get("symbol", "") self.log.add( f"EXIT_{reason}", f"P&L ₹{pnl:+.2f}", symbol=sym, data=action, ) self.total_pnl += pnl # Update risk guard loss tracker risk_guard.on_trade_closed(pnl) # ── SOURCE OF TRUTH: emit ONE close event to TSM (REAL only). # Updates authoritative positions/capital/realized-PnL AND # writes the trade to the journal (trade_history) — the # autonomous engine previously never journaled its trades. if self.config.mode == "REAL": try: trading_state_manager.apply_event(TradingEvent( event_type=EventType.POSITION_CLOSED, symbol=sym, broker=action.get("broker", ""), exchange=action.get("exchange", "NSE"), side=action.get("side", "BUY"), filled_quantity=int(action.get("qty", 0) or 0), avg_price=float(action.get("exit_price", 0.0) or 0.0), pnl=pnl, reason=reason, style=action.get("style", pool.style), strategy=action.get("strategy", ""), mode="REAL", source="autonomous_engine", )) except Exception as _tsm_err: logger.error(f"[Engine] TSM close event failed for {sym}: {_tsm_err}") # ── CRITICAL: flatten at the broker (OCO-safe) ─────────── # A broker-native STOPLOSS_MARKET order may be outstanding. # To avoid a DOUBLE exit (which would open an unintended # opposite position), we cancel that SL first and use the # cancel result as the arbiter: # • cancel succeeds → SL was still live → position still # open at broker → send the market exit order. # • cancel fails → SL already triggered/filled → # position already flat → do NOT send a second order; # the periodic reconcile confirms the close. # If no SL order was placed, software is the sole protection # and we send the exit unconditionally (prior behaviour). if self.config.mode == "REAL": try: from app.api.broker_api import manager as _bm pos_broker = action.get("broker", _bm.broker_name or "") broker_inst = _bm.get_broker(pos_broker) if pos_broker else _bm.broker sl_id = action.get("sl_order_id", "") send_exit = True if sl_id and broker_inst: try: cancel_res = broker_inst.cancel_order(sl_id) except Exception as _ce: cancel_res = {"success": False, "message": str(_ce)} if cancel_res.get("success"): self.log.add("SL_CANCELLED", f"Broker SL {sl_id} cancelled on {reason} — sending exit", symbol=sym) else: send_exit = False self.log.add("SL_ALREADY_FIRED", f"Broker SL {sl_id} not cancellable ({cancel_res.get('message','')}) — " f"assuming it already flattened the position; skipping exit order", symbol=sym) if send_exit and broker_inst and getattr(broker_inst, "_connected", False): exit_side = "SELL" if action.get("side", "BUY") == "BUY" else "BUY" exit_order = { "tradingsymbol": sym, "exchange": action.get("exchange", "NSE"), "transactiontype": exit_side, "ordertype": "MARKET", "quantity": action.get("qty", 1), "producttype": "INTRADAY", "price": 0, "duration": "DAY", "variety": "NORMAL", } exit_result = broker_inst.place_order(exit_order) if exit_result.get("success"): self.log.add("EXIT_ORDER_SENT", f"Broker exit order placed: {exit_result.get('order_id','')}", symbol=sym) else: self.log.add("EXIT_ORDER_FAIL", f"Broker exit FAILED: {exit_result.get('message','')}", symbol=sym) except Exception as _ex: self.log.add("EXIT_ORDER_ERR", f"Exit order exception: {_ex}", symbol=sym) # Record to learning DB if self.config.learning_enabled: try: learning_service.record_trade({ "symbol": sym, "style": action.get("style", pool.style), "broker": action.get("broker", ""), "side": action.get("side", "BUY"), "qty": action.get("qty", 0), "entry_price": action.get("entry_price", 0.0), "exit_price": action.get("exit_price", 0.0), "pnl": pnl, "pnl_pct": action.get("pnl_pct", 0.0), "exit_time": time.strftime("%Y-%m-%d %H:%M:%S"), "exit_reason": reason, "strategy": action.get("strategy", ""), "mode": self.config.mode, }) except Exception as ex: logger.warning(f"[Engine] Learning record failed: {ex}") # ── Helpers ─────────────────────────────────────────────────────────────── def _is_trading_hours(self) -> bool: """Check if current IST time is within configured trading window.""" # In PAPER mode always return True for testing if self.config.mode == "PAPER": return True try: now = datetime.now() sh, sm = map(int, self.config.trading_start.split(":")) eh, em = map(int, self.config.trading_end.split(":")) start_mins = sh * 60 + sm end_mins = eh * 60 + em now_mins = now.hour * 60 + now.minute return start_mins <= now_mins <= end_mins except Exception: return True def _map_style(self, strategy: str) -> str: """Map strategy name to capital pool style.""" mapping = { "SUPERTREND": "INTRADAY", "EMA_CROSSOVER": "INTRADAY", "VWAP": "SCALPING", "RSI_REVERSAL": "SWING", "MACD": "SWING", "AI": "SWING", } return mapping.get(strategy, "INTRADAY") # ── Singleton ───────────────────────────────────────────────────────────────── autonomous_engine = AutonomousEngine()