"""Sigil Trading Bot — Main orchestrator. Single-process async Python bot that runs all trading subsystems in coordinated event loops. """ from __future__ import annotations import asyncio from decimal import Decimal import os import sys import time from pathlib import Path import structlog from aiohttp import web from apscheduler.schedulers.asyncio import AsyncIOScheduler from prometheus_client import generate_latest, CONTENT_TYPE_LATEST # Add project root to path for imports sys.path.insert(0, str(Path(__file__).parent.parent)) from src.config import SigilConfig, load_config from src.contracts import AlertEvent from src.lifecycle import Lifecycle from src.observability.dashboard import handle_dashboard from src.observability.health import create_health_app from src.observability.logging import setup_logging from src.observability.metrics import ( sigil_info, sigil_db_size, sigil_signal_confidence, sigil_portfolio_value, sigil_snapshot_writes_total, ) from src.storage.database import write_heartbeat, run_retention_cleanup, get_pool log = structlog.get_logger() def _log_alert(event: AlertEvent) -> None: """Log alert events via structlog (replaces Discord AlertQueue).""" level_map = {"info": log.info, "warn": log.warning, "critical": log.critical} logger = level_map.get(event.severity, log.info) logger( "alert_event", event_type=event.event_type, symbol=event.symbol, message=event.message, mode=event.mode, ) class SigilBot: """Main orchestrator — wires all subsystems together.""" def __init__(self, config: SigilConfig) -> None: self.config = config self.lifecycle = Lifecycle(config) self.scheduler = AsyncIOScheduler() self._http_runner: web.AppRunner | None = None # Trading components — set during run() self._market_feed = None self._signal_engine = None self._pump_detector = None self._order_manager = None self._risk_manager = None self._position_tracker = None self._db_conn = None self._micro_manager = None async def run(self) -> None: """Main entry point — startup, run, shutdown.""" loop = asyncio.get_event_loop() self.lifecycle.setup_signal_handlers(loop) try: # Startup state machine components = await self.lifecycle.startup() self._db_conn = components["db"] # Start HTTP server (health + metrics + admin) await self._start_http_server() # Set bot info metric sigil_info.info( { "mode": self.config.mode, "exchange": self.config.exchange.exchange_id, "version": "0.1.0", "variant": os.environ.get("SIGIL_VARIANT", "baseline"), } ) # Initialise trading components trading_ready = await self._init_trading_components(components) # Grid stagger: offset scheduler start to avoid API burst startup_delay = int(os.environ.get("SIGIL_STARTUP_DELAY", "0")) if startup_delay: log.info("grid_stagger_delay", seconds=startup_delay) await asyncio.sleep(startup_delay) # Schedule ALL jobs (housekeeping + trading) self._schedule_jobs() self.scheduler.start() log.info("bot_started", mode=self.config.mode, trading=trading_ready) log.info( "orchestrator_running", mode=self.config.mode, trading=trading_ready ) # Main loop — wait for shutdown signal while not self.lifecycle.shutdown_event.is_set(): try: await asyncio.wait_for( self.lifecycle.shutdown_event.wait(), timeout=1.0, ) except asyncio.TimeoutError: pass except Exception as e: log.critical("bot_error", error=str(e)) raise finally: self.scheduler.shutdown(wait=False) if self._market_feed: await self._market_feed.stop() if self._http_runner: await self._http_runner.cleanup() await self.lifecycle.shutdown() async def _init_trading_components(self, components: dict) -> bool: """Initialise all trading components. Returns True if successful.""" assert self._db_conn is not None try: from src.observer.market_feed import MarketFeed from src.observer.pump_detector import PumpDetector from src.signals.engine import SignalEngine from src.executor.order_manager import OrderManager from src.executor.risk_manager import RiskManager from src.executor.position_tracker import PositionTracker from src.executor.draft_trader import DraftTrader exchange = self._create_exchange() self._market_feed = MarketFeed(self.config) self._pump_detector = PumpDetector(self.config, self._market_feed) draft_trader = DraftTrader(self.config, self._db_conn) self._position_tracker = PositionTracker( self.config, exchange, self._db_conn, _log_alert ) self._risk_manager = RiskManager( self.config, self._position_tracker, self._db_conn, _log_alert ) self._order_manager = OrderManager( self.config, exchange, self._db_conn, draft_trader, _log_alert, market_feed=self._market_feed, ) self._signal_engine = SignalEngine( self.config, self._market_feed, self._db_conn, ml_strategy=components.get("ml_strategy"), ) # Start market data feeds await self._market_feed.start() log.info("market_feed_started") # Fetch initial candle history for core symbols for symbol in self.config.core.symbols: await self._market_feed.fetch_candles(symbol, "1h", 100) log.info("initial_candles_fetched", symbols=self.config.core.symbols) # Micro-trading integration if self.config.micro.enabled: from src.micro_trading import MicroTradingManager self._micro_manager = MicroTradingManager( self.config, exchange, self._db_conn, self._market_feed, _log_alert ) log.info("micro_trading_manager_ready") return True except Exception as e: log.warning( "trading_init_failed", error=str(e), note="Running in minimal mode — no trading", ) return False # ------------------------------------------------------------------ # Trading Loop Jobs # ------------------------------------------------------------------ async def _core_strategy_job(self) -> None: """Evaluate core portfolio signals every 15 minutes.""" if not self._signal_engine or not self._risk_manager or not self._order_manager: return assert self._market_feed is not None for symbol in self.config.core.symbols: try: # Refresh candle data await self._market_feed.fetch_candles(symbol, "1h", 100) # Generate signals signals = await self._signal_engine.generate_signals(symbol) for signal in signals: if signal.direction == "hold": continue sigil_signal_confidence.labels(tier="core").observe( signal.confidence ) # Risk check decision = self._risk_manager.evaluate(signal) if not decision.approved: log.info( "signal_rejected", symbol=symbol, direction=signal.direction, confidence=signal.confidence, reason=decision.rejection_reason, step=decision.rejection_step, ) continue # Place order (draft or live) result = await self._order_manager.place_order(signal, decision) # Update position tracker so snapshots reflect the fill. # filled_qty is a portfolio %, convert to coin quantity. if ( result.filled_qty > 0 and self._position_tracker and result.avg_price > 0 ): starting = Decimal(str(self.config.core.draft_starting_capital)) usd_notional = result.filled_qty / Decimal("100") * starting coin_qty = usd_notional / result.avg_price self._position_tracker.update_position( symbol=symbol, mode=self.config.mode, quantity=coin_qty, avg_entry_price=result.avg_price, strategy_tier=signal.strategy_tier, ) log.info( "trade_executed", symbol=symbol, direction=signal.direction, confidence=signal.confidence, status=result.status.value, mode=result.mode, filled_qty=str(result.filled_qty), avg_price=str(result.avg_price), ) except Exception as e: log.error("core_strategy_error", symbol=symbol, error=str(e)) async def _pump_scan_job(self) -> None: """Scan for pump-and-dump opportunities every 30 seconds.""" if not self._pump_detector or not self._risk_manager or not self._order_manager: return try: signals = await self._pump_detector.scan() for signal in signals: sigil_signal_confidence.labels(tier="pump").observe(signal.confidence) # Risk check decision = self._risk_manager.evaluate(signal) if not decision.approved: log.info( "pump_signal_rejected", symbol=signal.symbol, confidence=signal.confidence, reason=decision.rejection_reason, ) continue # Place order (draft or live) result = await self._order_manager.place_order(signal, decision) # Update position tracker so snapshots reflect the fill. # filled_qty is a portfolio %, convert to coin quantity. if ( result.filled_qty > 0 and self._position_tracker and result.avg_price > 0 ): starting = Decimal(str(self.config.core.draft_starting_capital)) usd_notional = result.filled_qty / Decimal("100") * starting coin_qty = usd_notional / result.avg_price self._position_tracker.update_position( symbol=signal.symbol, mode=self.config.mode, quantity=coin_qty, avg_entry_price=result.avg_price, strategy_tier=signal.strategy_tier, ) log.info( "pump_trade_executed", symbol=signal.symbol, confidence=signal.confidence, status=result.status.value, mode=result.mode, max_hold_seconds=signal.max_hold_seconds, ) except Exception as e: log.error("pump_scan_error", error=str(e)) def _snapshot_sync(self, _conn=None) -> None: """Synchronous snapshot computation — runs in a thread via asyncio.to_thread.""" import contextlib if self._position_tracker is None: log.warning("snapshot_skipped_no_position_tracker") return if _conn is not None: conn_ctx = contextlib.nullcontext(_conn) else: conn_ctx = get_pool().connection() with conn_ctx as conn: assert conn is not None open_positions = self._position_tracker.get_all_positions(self.config.mode) pos_value = sum(p.quantity * p.avg_entry_price for p in open_positions) starting_capital = Decimal(str(self.config.core.draft_starting_capital)) cash = starting_capital - sum( p.quantity * p.avg_entry_price for p in open_positions ) total = cash + pos_value conn.execute( "INSERT INTO portfolio_snapshots " "(total_value_usd, cash_balance, positions_value, mode, snapshot_at) " "VALUES (%s, %s, %s, %s, NOW())", (str(total), str(cash), str(pos_value), self.config.mode), ) conn.commit() sigil_portfolio_value.labels(mode=self.config.mode).set(float(total)) sigil_snapshot_writes_total.labels(mode=self.config.mode, status="ok").inc() log.info( "pipeline_trace", stage="snapshot", status="ok", total=str(total), mode=self.config.mode, ) async def _portfolio_snapshot_job(self) -> None: """Take a portfolio snapshot every 5 minutes.""" try: await asyncio.to_thread(self._snapshot_sync) except Exception: log.exception("portfolio_snapshot_job_failed") sigil_snapshot_writes_total.labels( mode=self.config.mode, status="error" ).inc() async def _daily_digest_job(self) -> None: """Send daily performance digest via email.""" if not self._db_conn or not self.config.email_to: return try: from src.observability.email_digest import ( build_digest_html, send_digest_email, ) html = build_digest_html(self._db_conn, self.config.mode) send_digest_email(html, self.config, self._db_conn) log.info("daily_digest_sent", mode=self.config.mode) except Exception as e: log.error("daily_digest_error", error=str(e)) async def _drift_check_job(self) -> None: """Check for model drift every hour; email alert if detected.""" if not self._db_conn or not self.config.ml.enabled: return try: from src.signals.drift_monitor import DriftMonitor from src.observability.email_digest import ( render_markdown_email, send_digest_email, ) monitor = DriftMonitor(self._db_conn) accuracy = monitor.get_rolling_accuracy(window_hours=24, tier="core") drift_detected = monitor.check_drift() if drift_detected and self.config.smtp_host and self.config.email_to: html = render_markdown_email( "Sigil Drift Alert", f"""# Model Drift Detected **24h rolling accuracy has dropped below threshold.** - **Accuracy:** {accuracy:.1%} - **Threshold:** 50% - **Action required:** Emergency retrain recommended Check signal logs and recent market conditions. The model may need retraining on more recent data. """, accent_color="#ef4444", footer="Sigil — Vanlint Homelab", ) # Reuse send_digest_email for delivery (records to DB) send_digest_email(html, self.config, self._db_conn) log.warning( "drift_alert_emailed", accuracy_24h=round(accuracy, 4), ) except Exception as e: log.error("drift_check_error", error=str(e)) # ------------------------------------------------------------------ # Scheduling # ------------------------------------------------------------------ def _schedule_jobs(self) -> None: """Schedule all recurring jobs — housekeeping + trading.""" assert self._db_conn is not None db_conn = self._db_conn # --- Housekeeping jobs --- async def heartbeat_job(): write_heartbeat(self.config.heartbeat_path) async def retention_job(): results = run_retention_cleanup(db_conn) if any(v > 0 for v in results.values()): log.info("retention_cleanup", results=results) async def db_size_job(): try: row = db_conn.execute( "SELECT pg_database_size(current_database()) AS size" ).fetchone() if row: sigil_db_size.set(row["size"]) except Exception: pass self.scheduler.add_job(heartbeat_job, "interval", seconds=30, id="heartbeat") self.scheduler.add_job( retention_job, "cron", day_of_week="sun", hour=3, id="retention" ) self.scheduler.add_job(db_size_job, "interval", minutes=1, id="db_size") # --- Trading loop jobs --- # Core strategy evaluation every 15 minutes if self.config.core.symbols: self.scheduler.add_job( self._core_strategy_job, "interval", minutes=self.config.core.signal_interval_minutes, id="core_strategy", max_instances=1, misfire_grace_time=60, ) # Pump-and-dump scan every 30 seconds if self.config.pump.max_total_exposure_pct > 0: self.scheduler.add_job( self._pump_scan_job, "interval", seconds=self.config.pump.scan_interval_seconds, id="pump_scan", max_instances=1, ) # Portfolio snapshot every 5 minutes self.scheduler.add_job( self._portfolio_snapshot_job, "interval", minutes=self.config.core.reconciliation_interval_minutes, id="portfolio_snapshot", max_instances=1, ) # Daily digest at 08:00 ACDT self.scheduler.add_job( self._daily_digest_job, "cron", hour=8, minute=0, id="daily_digest", misfire_grace_time=300, ) # Drift check every hour (only runs when ML is enabled) if self.config.ml.enabled: self.scheduler.add_job( self._drift_check_job, "interval", hours=1, id="drift_check", max_instances=1, misfire_grace_time=120, ) # Micro-trading scan and exit jobs if self.config.micro.enabled and self._micro_manager is not None: self.scheduler.add_job( self._micro_manager.scan_and_trade, "interval", seconds=60, id="micro_scan", max_instances=1, ) self.scheduler.add_job( self._micro_manager.manage_exits, "interval", seconds=30, id="micro_exits", max_instances=1, ) log.info("micro_jobs_scheduled") log.info( "jobs_scheduled", core_interval_min=self.config.core.signal_interval_minutes, pump_interval_sec=self.config.pump.scan_interval_seconds, ) # ------------------------------------------------------------------ # Infrastructure # ------------------------------------------------------------------ def _create_exchange(self): """Create ccxt exchange instance.""" import ccxt exchange_class = getattr(ccxt, self.config.exchange.exchange_id, None) if exchange_class is None: raise ValueError(f"Unknown exchange: {self.config.exchange.exchange_id}") exchange = exchange_class( { "apiKey": self.config.exchange.api_key, "secret": self.config.exchange.api_secret, "enableRateLimit": True, "options": {"defaultType": "spot"}, } ) if self.config.mode == "draft": log.info( "exchange_draft_mode", note="orders routed to DraftTrader, exchange used for market data only", ) return exchange async def _start_http_server(self) -> None: """Start aiohttp server for health, metrics, and admin endpoints.""" app = create_health_app(self.config) async def handle_metrics(request: web.Request) -> web.Response: return web.Response( body=generate_latest(), content_type=CONTENT_TYPE_LATEST, ) async def handle_kill(request: web.Request) -> web.Response: import hmac as hmac_mod auth = request.headers.get("X-Admin-Key", "") if not self.config.admin_api_key: return web.json_response( {"error": "admin key not configured"}, status=503 ) if not hmac_mod.compare_digest(auth, self.config.admin_api_key): return web.json_response({"error": "unauthorized"}, status=403) kill_path = Path(self.config.risk.kill_switch_path) kill_path.parent.mkdir(parents=True, exist_ok=True) kill_path.write_text(f"Killed via API at {time.time()}\n") log.critical("kill_switch_activated", method="api") return web.json_response({"status": "halted", "timestamp": time.time()}) app.router.add_get("/metrics", handle_metrics) app.router.add_get("/dashboard", handle_dashboard) app.router.add_post("/admin/kill", handle_kill) self._http_runner = web.AppRunner(app) await self._http_runner.setup() site = web.TCPSite( self._http_runner, self.config.risk.admin_api_bind, self.config.risk.admin_api_port, ) await site.start() log.info( "http_server_started", bind=self.config.risk.admin_api_bind, port=self.config.risk.admin_api_port, ) def main() -> None: """Entry point.""" try: from dotenv import load_dotenv env_file = Path("secrets.env") if env_file.exists(): load_dotenv(env_file) except ImportError: pass setup_logging( log_dir=os.getenv("SIGIL_LOG_DIR", "/app/data/logs"), level=os.getenv("LOG_LEVEL", "INFO"), ) config = load_config( config_dir=os.getenv("SIGIL_CONFIG_DIR", "config"), ) bot = SigilBot(config) asyncio.run(bot.run()) if __name__ == "__main__": main()