""" AkDevz AI Market Intelligence — Autonomous Continuous Trading & Research Daemon Operates 24/7 on VPS / Dedicated Server: - Market Hours (09:15-15:30 IST): High-frequency real-time scanning, risk gating, execution & trailing SL management - Post Market (15:30-23:59 IST): Trade reconciliation, outcome analysis, next-day watchlists - Weekends: Strategy validation & research memory aggregation """ import sys from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) import time import signal import logging import argparse from datetime import datetime, time as dtime from zoneinfo import ZoneInfo import pymysql from python.config.settings import settings from python.data.market_feed import MarketFeed from python.data.data_normalizer import DataNormalizer from python.analytics.technical_engine import TechnicalEngine from python.analytics.market_regime import MarketRegimeEngine from python.analytics.support_resistance import SupportResistanceEngine from python.trading.execution_engine import ExecutionEngine IST = ZoneInfo("Asia/Kolkata") logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] [ContinuousDaemon] %(message)s" ) logger = logging.getLogger("ContinuousDaemon") def get_scan_universe() -> list: """Dynamically queries all active assets from DB, falling back to top leaders.""" try: conn = pymysql.connect( host=settings.db_host, port=settings.db_port, user=settings.db_username, password=settings.db_password, database=settings.db_database, connect_timeout=3 ) with conn.cursor() as cur: cur.execute("SELECT symbol FROM assets WHERE is_active = 1 ORDER BY segment DESC, id ASC") rows = cur.fetchall() if rows: conn.close() return [r[0] for r in rows] conn.close() except Exception as e: logger.warning(f"Could not load scan universe from DB: {e}") return [ "NSE:NIFTY50-INDEX", "NSE:BANKNIFTY-INDEX", "NSE:FINNIFTY-INDEX", "NSE:RELIANCE-EQ", "NSE:HDFCBANK-EQ", "NSE:ICICIBANK-EQ", "NSE:INFY-EQ", "NSE:TCS-EQ" ] class ContinuousTradingWorker: def __init__(self, mode: str = "paper", force: bool = False): self.mode = mode self.force = force self.running = True self.engine = ExecutionEngine() self.universe = get_scan_universe() self.setup_signal_handlers() def setup_signal_handlers(self): signal.signal(signal.SIGINT, self._handle_exit) signal.signal(signal.SIGTERM, self._handle_exit) def _handle_exit(self, signum, frame): logger.info(f"Received shutdown signal ({signum}). Gracefully stopping daemon...") self.running = False def is_market_hours(self, dt: datetime) -> bool: if self.force: return True if dt.weekday() >= 5: # Saturday or Sunday return False cur_time = dt.time() market_open = dtime(9, 15) market_close = dtime(15, 30) return market_open <= cur_time <= market_close def run_market_hours_cycle(self): logger.info(f"Executing active market scan cycle across {len(self.universe)} instruments...") for symbol in self.universe[:25]: # Scan liquid top 25 leaders try: # 1. Fetch real market candles raw_candles = MarketFeed.get_real_candles(symbol, count=60) if not raw_candles or len(raw_candles) < 10: continue norm_candles = DataNormalizer.normalize_candles(raw_candles) np_data = DataNormalizer.to_numpy_arrays(norm_candles) c = np_data["close"] h = np_data["high"] l = np_data["low"] v = np_data["volume"] ltp = float(c[-1]) vwap_arr = TechnicalEngine.calculate_vwap(h, l, c, v) vwap = float(vwap_arr[-1]) if not np.isnan(vwap_arr[-1]) else ltp ema9 = float(TechnicalEngine.calculate_ema(c, 9)[-1]) ema20 = float(TechnicalEngine.calculate_ema(c, 20)[-1]) atr14 = float(TechnicalEngine.calculate_atr(h, l, c, 14)[-1]) pivots = SupportResistanceEngine.calculate_daily_pivots( high=float(max(h[-50:])), low=float(min(l[-50:])), close=ltp ) # Classify regime regime = MarketRegimeEngine.classify_regime( ltp=ltp, vwap=vwap, ema_9=ema9, ema_20=ema20, atr_14=atr14, rvol=1.0, bb_bandwidth=0.35, squeeze_state="NORMAL" ) # 2. Check for Breakout Setup (Bullish above R1 or Bearish below S1) decision = "NO_TRADE" side = None trigger_price = None sl = None t1 = None t2 = None r1 = pivots.get("r1", ltp * 1.005) s1 = pivots.get("s1", ltp * 0.995) if ltp > r1 and ema9 > ema20 and ltp > vwap: decision = "BUY_CE" side = "BUY" trigger_price = ltp sl = max(vwap, ltp - (1.5 * atr14)) t1 = ltp + (2.0 * (ltp - sl)) t2 = ltp + (3.0 * (ltp - sl)) elif ltp < s1 and ema9 < ema20 and ltp < vwap: decision = "BUY_PE" side = "SELL" trigger_price = ltp sl = min(vwap, ltp + (1.5 * atr14)) t1 = ltp - (2.0 * (sl - ltp)) t2 = ltp - (3.0 * (sl - ltp)) if decision in ["BUY_CE", "BUY_PE"] and side: candidate = { "symbol": symbol, "strategy_code": f"{symbol}_{decision}", "side": side, "trigger_price": trigger_price, "stop_loss": round(sl, 2), "target_1": round(t1, 2), "target_2": round(t2, 2), "risk_reward": 2.0 } exec_res = self.engine.execute_candidate( candidate=candidate, mode=self.mode, confidence_score=78.0 ) logger.info(f"Scan Trigger for {symbol} ({decision}): {exec_res.get('status')} - {exec_res.get('reason', '')}") except Exception as e: logger.error(f"Error scanning {symbol}: {e}") # 3. Mark open positions to market and evaluate trailing SL try: self.monitor_open_positions() except Exception as e: logger.error(f"Error in position monitor: {e}") def monitor_open_positions(self): try: self.engine.manage_open_positions() except Exception as e: logger.error(f"Error managing open positions: {e}") def run_post_market_cycle(self): logger.info("Executing post-market reconciliation and performance audit...") conn = self.engine.get_connection() if not conn: return try: with conn.cursor() as cur: cur.execute(""" SELECT COUNT(*) as total_closed_positions, COALESCE(SUM(realized_pnl), 0.0) as net_pnl, SUM(CASE WHEN realized_pnl > 0 THEN 1 ELSE 0 END) as wins FROM trading_positions WHERE status = 'CLOSED' AND DATE(closed_at) = CURDATE() """) stats = cur.fetchone() or {} logger.info(f"Session Summary: {stats.get('total_closed_positions', 0)} closed positions, Net Realized P&L: ₹{stats.get('net_pnl', 0)}") finally: conn.close() def start(self, poll_interval_sec: int = 20, run_once: bool = False): logger.info(f"Continuous Trading Worker started in {self.mode.upper()} mode. Poll interval: {poll_interval_sec}s") while self.running: now = datetime.now(IST) if self.is_market_hours(now): self.run_market_hours_cycle() else: logger.info(f"Market Closed ({now.strftime('%H:%M:%S IST')}). Running post-market / standby routines...") self.run_post_market_cycle() if run_once: break time.sleep(poll_interval_sec) logger.info("Continuous Trading Worker stopped cleanly.") def main(): parser = argparse.ArgumentParser(description="AkDevz Continuous Trading Daemon") parser.add_argument("--mode", type=str, default="paper", choices=["paper", "live"]) parser.add_argument("--interval", type=int, default=20) parser.add_argument("--once", action="store_true", help="Run single cycle and exit") parser.add_argument("--force", action="store_true", help="Force active market scan regardless of market hours") args = parser.parse_args() worker = ContinuousTradingWorker(mode=args.mode, force=args.force) worker.start(poll_interval_sec=args.interval, run_once=args.once) if __name__ == "__main__": main()