""" ========================================= CV6 AI Trading OS — F&O Engine ========================================= Real F&O trading engine: FLOW: 1. OI Analysis → market direction 2. AI Consensus → confirm direction 3. Strike Selection → ATM / OTM 4. Black-Scholes → fair value check 5. Angel One NFO → place real order 6. Position Monitor → premium-based SL Supports: • NIFTY weekly options • BANKNIFTY weekly options • CE / PE buying (Phase 1) ========================================= """ from __future__ import annotations import time import uuid from dataclasses import dataclass, field, asdict from datetime import date, datetime from typing import Any, Dict, List, Optional from loguru import logger from app.fno.fno_symbols import fno_symbols, OptionContract, LOT_SIZES from app.fno.options_pricing import price_option, calc_iv, greeks_summary, OptionGreeks # ── F&O Trade record ────────────────────────────────────────────────────────── @dataclass class FnoTrade: trade_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8]) underlying: str = "" option_type: str = "" strike: float = 0.0 expiry: str = "" symbol: str = "" token: str = "" exchange: str = "NFO" lots: int = 1 lot_size: int = 75 quantity: int = 0 entry_premium: float = 0.0 entry_cost: float = 0.0 entry_time: str = "" order_id: str = "" sl_premium: float = 0.0 target_premium: float = 0.0 status: str = "OPEN" exit_premium: float = 0.0 exit_time: str = "" exit_order_id: str = "" pnl: float = 0.0 pnl_pct: float = 0.0 charges: float = 0.0 delta_at_entry: float = 0.0 theta_at_entry: float = 0.0 iv_at_entry: float = 0.0 ai_signal: str = "" ai_confidence: float = 0.0 ai_reason: str = "" def to_dict(self) -> dict: return asdict(self) # ── F&O charges (NSE NFO) ───────────────────────────────────────────────────── def calc_fno_charges(premium: float, quantity: int, side: str) -> float: turnover = premium * quantity brokerage = 20.0 stt = turnover * 0.000625 if side == "SELL" else 0.0 exchange = turnover * 0.0000345 gst = (brokerage + exchange) * 0.18 sebi = turnover * 0.000001 stamp = turnover * 0.00003 if side == "BUY" else 0.0 return round(brokerage + stt + exchange + gst + sebi + stamp, 2) # ── Main F&O Engine ─────────────────────────────────────────────────────────── class FnoEngine: def __init__(self): self._open_trades: Dict[str, FnoTrade] = {} def analyse_direction(self, underlying: str = "NIFTY", use_ai: bool = True) -> Dict[str, Any]: result = { "direction": "NO_TRADE", "confidence": 0.0, "reason": "", "oi_signal": "N/A", "pcr": 0.0, "ai_signal": "N/A", "ai_confidence": 0.0, "ai_reason": "", } try: from app.market.nse_data import get_option_chain_oi, get_nifty_change oi = get_option_chain_oi(underlying) if oi: result["oi_signal"] = oi["oi_signal"] result["pcr"] = oi["pcr"] nifty_chg = get_nifty_change() or 0.0 except Exception as e: logger.warning(f"[FnO] OI fetch: {e}") nifty_chg = 0.0 ai_direction = "HOLD" if use_ai: try: import asyncio from app.ai.ai_consensus import consensus_engine loop = asyncio.new_event_loop() market_data = {"ltp": 0, "change_pct": nifty_chg, "oi_signal": result["oi_signal"], "pcr": result["pcr"]} cr = loop.run_until_complete( consensus_engine.consensus(underlying, market_data, timeout=10) ) loop.close() ai_direction = cr.signal result["ai_signal"] = cr.signal result["ai_confidence"] = cr.confidence result["ai_reason"] = cr.reason except Exception as e: logger.warning(f"[FnO] AI: {e}") oi_bull = result["oi_signal"] == "BULLISH" oi_bear = result["oi_signal"] == "BEARISH" ai_buy = ai_direction == "BUY" ai_sell = ai_direction == "SELL" if oi_bull and (ai_buy or not use_ai): result.update(direction="BUY_CE", confidence=0.75 if ai_buy else 0.55, reason=f"PCR={result['pcr']:.2f} bullish + AI BUY") elif oi_bear and (ai_sell or not use_ai): result.update(direction="BUY_PE", confidence=0.75 if ai_sell else 0.55, reason=f"PCR={result['pcr']:.2f} bearish + AI SELL") elif oi_bull and nifty_chg > 0.3: result.update(direction="BUY_CE", confidence=0.50, reason=f"PCR bullish + Nifty +{nifty_chg:.2f}%") elif oi_bear and nifty_chg < -0.3: result.update(direction="BUY_PE", confidence=0.50, reason=f"PCR bearish + Nifty {nifty_chg:.2f}%") else: result.update(direction="NO_TRADE", reason="OI and AI signals not aligned") return result def place_trade( self, underlying: str, spot_price: float, option_type: str, lots: int = 1, strike: Optional[float] = None, strike_style: str = "OTM1", sl_pct: float = 50.0, target_pct: float = 100.0, expiry_date: Optional[date] = None, broker_name: str = "angelone", paper: bool = False, iv_pct: float = 15.0, ) -> Dict[str, Any]: ot = option_type.upper() if strike is None: step = 100 if "BANK" in underlying.upper() else 50 atm = round(round(spot_price / step) * step, 0) steps = 1 if strike_style == "OTM1" else (2 if strike_style == "OTM2" else 0) strike = atm + (step * steps if ot == "CE" else -step * steps) contract = fno_symbols.resolve(underlying, strike, ot, expiry_date) if contract is None: # Fallback: try to auto-load scrip master and retry once fno_symbols.load(force=True) contract = fno_symbols.resolve(underlying, strike, ot, expiry_date) if contract is None: return {"success": False, "message": f"Symbol not found: {underlying} {strike} {ot}. " f"Scrip master may not have this expiry yet."} lot_size = contract.lot_size quantity = lots * lot_size today = date.today() dte = max((contract.expiry_date - today).days, 0) greeks = price_option(spot_price, strike, dte, iv_pct, ot) fill_premium = greeks.fair_value order_id = f"PAPER-{uuid.uuid4().hex[:8].upper()}" if not paper: try: from app.broker.broker_factory import BrokerFactory broker = BrokerFactory.get_broker(broker_name) conn = broker.connect() if not conn.get("success"): return {"success": False, "message": f"Broker: {conn.get('message')}"} payload = { "variety": "NORMAL", "tradingsymbol": contract.symbol, "symboltoken": contract.token, "transactiontype": "BUY", "exchange": contract.exchange, "ordertype": "MARKET", "producttype": "CARRYFORWARD", "duration": "DAY", "price": "0", "squareoff": "0", "stoploss": "0", "quantity": str(quantity), } res = broker.place_order(payload) if not res.get("success"): return {"success": False, "message": f"Order: {res.get('message')}"} order_id = str(res.get("order_id", order_id)) except Exception as e: logger.error(f"[FnO] Place order error: {e}") return {"success": False, "message": str(e)} sl_premium = round(fill_premium * (1 - sl_pct / 100), 2) target_premium = round(fill_premium * (1 + target_pct / 100), 2) charges = calc_fno_charges(fill_premium, quantity, "BUY") trade = FnoTrade( underlying=underlying.upper(), option_type=ot, strike=strike, expiry=contract.expiry, symbol=contract.symbol, token=contract.token, exchange=contract.exchange, lots=lots, lot_size=lot_size, quantity=quantity, entry_premium=fill_premium, entry_cost=fill_premium * quantity + charges, entry_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), order_id=order_id, sl_premium=sl_premium, target_premium=target_premium, charges=charges, delta_at_entry=greeks.delta, theta_at_entry=greeks.theta, iv_at_entry=greeks.iv, ) self._open_trades[trade.trade_id] = trade logger.info( f"[FnO] {'PAPER' if paper else 'REAL'} trade placed: " f"{contract.symbol} | {lots}L × {lot_size} | " f"₹{fill_premium:.2f}/unit | Cost ₹{trade.entry_cost:.0f} | " f"SL ₹{sl_premium} | Target ₹{target_premium}" ) return { "success": True, "trade_id": trade.trade_id, "order_id": order_id, "symbol": contract.symbol, "strike": strike, "option_type": ot, "lots": lots, "quantity": quantity, "entry_premium": fill_premium, "entry_cost": trade.entry_cost, "sl_premium": sl_premium, "target_premium": target_premium, "days_to_expiry": dte, "charges": charges, "paper_trade": paper, "greeks": { "fair_value": greeks.fair_value, "delta": greeks.delta, "theta": greeks.theta, "vega": greeks.vega, "iv": greeks.iv, "moneyness": greeks.moneyness, }, } def close_trade(self, trade_id: str, exit_premium: float, broker_name: str = "angelone", paper: bool = False) -> Dict: trade = self._open_trades.get(trade_id) if not trade: return {"success": False, "message": f"Trade {trade_id} not found"} exit_order_id = f"EXIT-{uuid.uuid4().hex[:8].upper()}" if not paper and not trade.order_id.startswith("PAPER"): try: from app.broker.broker_factory import BrokerFactory broker = BrokerFactory.get_broker(broker_name) broker.connect() res = broker.place_order({ "variety": "NORMAL", "tradingsymbol": trade.symbol, "symboltoken": trade.token, "transactiontype": "SELL", "exchange": trade.exchange, "ordertype": "MARKET", "producttype": "CARRYFORWARD", "duration": "DAY", "price": "0", "squareoff": "0", "stoploss": "0", "quantity": str(trade.quantity), }) exit_order_id = str(res.get("order_id", exit_order_id)) except Exception as e: logger.error(f"[FnO] Exit error: {e}") exit_charges = calc_fno_charges(exit_premium, trade.quantity, "SELL") gross_pnl = (exit_premium - trade.entry_premium) * trade.quantity net_pnl = gross_pnl - trade.charges - exit_charges pnl_pct = (net_pnl / trade.entry_cost * 100) if trade.entry_cost > 0 else 0.0 trade.exit_premium = exit_premium trade.exit_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") trade.exit_order_id = exit_order_id trade.status = "MANUAL_EXIT" trade.pnl = round(net_pnl, 2) trade.pnl_pct = round(pnl_pct, 2) trade.charges += exit_charges logger.info(f"[FnO] Closed {trade.symbol} | P&L ₹{net_pnl:+.2f} ({pnl_pct:+.1f}%)") return {"success": True, "trade_id": trade_id, "pnl": round(net_pnl, 2), "pnl_pct": round(pnl_pct, 2)} def monitor_positions(self, current_premiums: Dict[str, float]) -> List[Dict]: actions = [] for trade in list(self._open_trades.values()): if trade.status != "OPEN": continue cp = current_premiums.get(trade.symbol) if cp is None: continue reason = None if cp <= trade.sl_premium: reason = "SL_HIT" elif cp >= trade.target_premium: reason = "TARGET_HIT" if reason: exit_charges = calc_fno_charges(cp, trade.quantity, "SELL") gross = (cp - trade.entry_premium) * trade.quantity net = gross - trade.charges - exit_charges pct = (net / trade.entry_cost * 100) if trade.entry_cost > 0 else 0.0 trade.exit_premium = cp trade.exit_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") trade.status = reason trade.pnl = round(net, 2) trade.pnl_pct = round(pct, 2) trade.charges += exit_charges actions.append({"trade_id": trade.trade_id, "symbol": trade.symbol, "reason": reason, "pnl": round(net, 2)}) return actions def open_positions(self) -> List[Dict]: return [t.to_dict() for t in self._open_trades.values() if t.status == "OPEN"] def all_trades(self) -> List[Dict]: return [t.to_dict() for t in self._open_trades.values()] def portfolio_pnl(self, current_premiums: Dict[str, float] = None) -> Dict: cp = current_premiums or {} invested = current_value = realised = 0.0 for t in self._open_trades.values(): if t.status == "OPEN": invested += t.entry_cost curr = cp.get(t.symbol, t.entry_premium) current_value += curr * t.quantity else: realised += t.pnl unrealised = current_value - invested return { "invested": round(invested, 2), "current_value": round(current_value, 2), "unrealised_pnl": round(unrealised, 2), "realised_pnl": round(realised, 2), "total_pnl": round(unrealised + realised, 2), } # ── Singleton ───────────────────────────────────────────────────────────────── fno_engine = FnoEngine()