main.py

"""
Pantheon OS — Main Entry Point

Two run modes:
  1. n8n webhook server (default) — n8n triggers ZEUS via HTTP POST
  2. Standalone loop — python main.py --standalone (for local testing)

Endpoints:
  POST /run    → one pipeline cycle
  POST /halt   → emergency halt
  GET  /status → portfolio + agent health state
  GET  /health → liveness check
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import threading
import time
from datetime import datetime, timedelta, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer

from dotenv import load_dotenv

from agents.zeus import ZeusConfig, ZeusOrchestrator
from config.settings import load_settings
from core.logging_setup import configure_logging

load_dotenv()   # loads .env file if present

logger = logging.getLogger("main")

_API_KEY = os.getenv("ZEUS_API_KEY")  # if unset → auth disabled (local dev)


def build_zeus() -> ZeusOrchestrator:
    settings = load_settings()
    config = ZeusConfig(
        max_portfolio_drawdown_pct = settings.get("max_drawdown_pct", 0.08),
        max_open_positions         = settings.get("max_open_positions", 10),
        paper_trading              = settings.get("paper_trading", True),
        mock_execution             = settings.get("mock_execution", True),
        use_llm_reasoning          = settings.get("use_llm_reasoning", True),
        # Single equity source — let ZeusOrchestrator resolve + fail-closed.
        # Pass None so the constructor reads settings and raises if absent,
        # rather than baking a wrong default here.
        default_account_equity     = settings.get("account_equity", settings.get("default_account_equity")),
        stop_loss_pct              = settings.get("stop_loss_pct", 0.03),
        take_profit_pct            = settings.get("take_profit_pct", 0.06),
    )
    return ZeusOrchestrator(config)


# ---------------------------------------------------------------------------
# n8n Webhook Server
# ---------------------------------------------------------------------------

# Single-process design: _zeus is module-level state shared between the HTTP
# handler threads and the auto-run daemon. This is intentional — the server
# runs as one process with workers=1. Do not scale to multiple workers without
# replacing this with a proper job queue (e.g. Celery + Redis).
_zeus: ZeusOrchestrator | None = None
_run_lock = threading.Lock()  # prevents concurrent run_once() calls


class ZeusHandler(BaseHTTPRequestHandler):
    def log_message(self, format, *args):
        logger.debug("HTTP %s", format % args)

    def _check_api_key(self) -> bool:
        if not _API_KEY:
            return True  # auth disabled in local dev
        provided = self.headers.get("X-API-Key", "")
        if provided != _API_KEY:
            self._json_response(401, {"error": "unauthorized"})
            return False
        return True

    def do_GET(self):
        if self.path == "/status":
            self._handle_status()
        elif self.path == "/health":
            self._json_response(200, {"status": "ok", "pipeline": _zeus.status.value if _zeus else "not started"})
        elif self.path == "/agents":
            self._handle_agents()
        else:
            self._json_response(404, {"error": "not found"})

    def do_POST(self):
        if not self._check_api_key():
            return
        if self.path == "/run":
            self._handle_run()
        elif self.path in ("/run/research", "/run/research/historical"):
            historical = self.path.endswith("/historical")
            self._handle_research(historical=historical)
        elif self.path == "/run/backtest":
            self._handle_backtest()
        elif self.path == "/run/replay":
            self._handle_replay()
        elif self.path == "/halt":
            self._handle_halt()
        elif self.path == "/resume":
            self._handle_resume()
        elif self.path == "/alert":
            self._handle_alert()
        elif self.path == "/admin/icarus/quality/reset":
            self._handle_icarus_quality_reset()
        else:
            self._json_response(404, {"error": "not found"})

    def _handle_run(self):
        try:
            if not _run_lock.acquire(blocking=False):
                self._json_response(409, {"error": "pipeline already running"})
                return
            try:
                runs = _zeus.run_once()
            finally:
                _run_lock.release()
            summary = [
                {
                    "run_id":      r.run_id,
                    "killed_at":   r.killed_at_stage,
                    "kill_reason": r.kill_reason,
                    "reasoning":   r.trace.zeus_reasoning if r.trace else None,
                    "trade": {
                        "symbol":   r.trade_result.symbol,
                        "side":     r.trade_result.side,
                        "order_id": r.trade_result.order_id,
                        "fill":     r.trade_result.fill_price,
                    } if r.trade_result and r.trade_result.symbol else None,
                }
                for r in runs
            ]
            self._json_response(200, {"pipeline_runs": summary, "count": len(runs)})
        except Exception as exc:
            logger.exception("[MAIN] /run failed")
            self._json_response(500, {"error": str(exc)})

    def _handle_research(self, historical: bool = False):
        try:
            summary = _zeus.run_research_cycle(historical=historical)
            self._json_response(200, {"status": "ok", "historical": historical, "research": summary})
        except Exception as exc:
            logger.exception("[MAIN] /run/research failed")
            self._json_response(500, {"error": str(exc)})

    def _handle_backtest(self):
        try:
            summary = _zeus.run_backtest()
            self._json_response(200, {"status": "ok", "backtest": summary})
        except Exception as exc:
            logger.exception("[MAIN] /run/backtest failed")
            self._json_response(500, {"error": str(exc)})

    def _handle_replay(self):
        try:
            summary = _zeus.run_replay()
            self._json_response(200, {"status": "ok", "replay": summary})
        except Exception as exc:
            logger.exception("[MAIN] /run/replay failed")
            self._json_response(500, {"error": str(exc)})

    def _handle_alert(self):
        """POST /alert — send a Telegram alert via Argus.
        Body: {"message": "...", "source": "..."} (source is optional label)
        Used by n8n VPS watchdog and any external monitor.
        """
        try:
            length = int(self.headers.get("Content-Length", 0))
            body = json.loads(self.rfile.read(length)) if length else {}
            message = body.get("message", "").strip()
            source  = body.get("source", "external")
            if not message:
                self._json_response(400, {"error": "message field required"})
                return
            full_msg = f"[{source}] {message}"
            _zeus.argus.send_alert(full_msg)
            logger.info("[MAIN] /alert sent from %s: %s", source, message[:80])
            self._json_response(200, {"status": "sent", "message": full_msg})
        except Exception as exc:
            logger.exception("[MAIN] /alert failed")
            self._json_response(500, {"error": str(exc)})

    def _handle_icarus_quality_reset(self):
        """POST /admin/icarus/quality/reset — clear a suppressed quality pattern.
        Body: {"pattern": "<hermes_type>:<keyword>"}. Lets an operator un-mute a
        signal pattern that the quality filter learned to suppress."""
        try:
            length  = int(self.headers.get("Content-Length", 0))
            body    = json.loads(self.rfile.read(length)) if length else {}
            pattern = (body.get("pattern") or "").strip()
            if not pattern:
                self._json_response(400, {"error": "pattern field required"})
                return
            deleted = _zeus.icarus.reset_quality_filter(pattern)
            self._json_response(200, {"status": "reset", "pattern": pattern, "keys_deleted": deleted})
        except Exception as exc:
            logger.exception("[MAIN] /admin/icarus/quality/reset failed")
            self._json_response(500, {"error": str(exc)})

    def _handle_halt(self):
        _zeus.halt(reason="n8n manual halt")
        self._json_response(200, {"status": "halted"})

    def _handle_resume(self):
        _zeus.resume()
        self._json_response(200, {"status": "running"})

    def _handle_status(self):
        state = _zeus.argus.portfolio_state()
        cb_status = _zeus.cb.status()
        self._json_response(200, {
            "pipeline_status":  _zeus.status.value,
            "open_positions":   _zeus.argus.open_position_count(),
            "equity":           state.total_equity,
            "drawdown_pct":     round(state.current_drawdown_pct * 100, 2),
            "paper_trading":    _zeus.config.paper_trading,
            "mock_execution":   _zeus.config.mock_execution,
            "circuit_breakers": cb_status,
            "seniority":        _zeus.get_seniority_report(),
        })

    def _handle_agents(self):
        reports = _zeus.get_health_reports()
        self._json_response(200, {
            "agents": [
                {
                    "name":    r.agent_name,
                    "status":  r.status.value,
                    "message": r.message,
                    "checked": r.checked_at.isoformat(),
                }
                for r in reports
            ]
        })

    def _json_response(self, code: int, body: dict):
        payload = json.dumps(body, default=str).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", len(payload))
        self.end_headers()
        self.wfile.write(payload)


def _edt_offset(utc_dt: datetime) -> int:
    """Return ET UTC offset: -4 (EDT) or -5 (EST).
    EDT starts 2nd Sunday of March, ends 1st Sunday of November.
    """
    y = utc_dt.year
    # 2nd Sunday of March
    march1_wd = datetime(y, 3, 1).weekday()          # 0=Mon … 6=Sun
    first_sun_mar = 1 + (6 - march1_wd) % 7
    dst_start = datetime(y, 3, first_sun_mar + 7, 2)  # +7 = second Sunday
    # 1st Sunday of November
    nov1_wd = datetime(y, 11, 1).weekday()
    first_sun_nov = 1 + (6 - nov1_wd) % 7
    dst_end = datetime(y, 11, first_sun_nov, 2)
    return -4 if dst_start <= utc_dt.replace(tzinfo=None) < dst_end else -5


def _cet_offset(utc_dt: datetime) -> int:
    """Return CET UTC offset: +2 (CEST, summer) or +1 (CET, winter).
    CEST starts last Sunday of March, ends last Sunday of October.
    """
    y = utc_dt.year
    # Last Sunday of March
    import calendar
    last_day_mar = calendar.monthrange(y, 3)[1]
    d = datetime(y, 3, last_day_mar)
    while d.weekday() != 6:
        d -= timedelta(days=1)
    dst_start = d.replace(hour=2)
    # Last Sunday of October
    last_day_oct = calendar.monthrange(y, 10)[1]
    d = datetime(y, 10, last_day_oct)
    while d.weekday() != 6:
        d -= timedelta(days=1)
    dst_end = d.replace(hour=3)
    return +2 if dst_start <= utc_dt.replace(tzinfo=None) < dst_end else +1


# NYSE holidays — fixed dates only (observed rules applied: if holiday falls
# on Saturday the prior Friday is observed; Sunday → following Monday).
# Extend this set each December for the coming year. No external dependency.
_NYSE_HOLIDAYS = {
    # 2025
    (2025, 1, 1), (2025, 1, 20), (2025, 2, 17), (2025, 4, 18),
    (2025, 5, 26), (2025, 6, 19), (2025, 7, 4), (2025, 9, 1),
    (2025, 11, 27), (2025, 12, 25),
    # 2026
    (2026, 1, 1), (2026, 1, 19), (2026, 2, 16), (2026, 4, 3),
    (2026, 5, 25), (2026, 6, 19), (2026, 7, 3), (2026, 9, 7),
    (2026, 11, 26), (2026, 12, 25),
}

# NYSE early-close days (13:00 ET close) — day before Independence Day,
# day after Thanksgiving, Christmas Eve when not a full holiday.
# Extend this set each December for the coming year. No external dependency.
_NYSE_EARLY_CLOSE = {
    # 2025
    (2025, 7, 3), (2025, 11, 28), (2025, 12, 24),
    # 2026
    (2026, 7, 2), (2026, 11, 27), (2026, 12, 24),
}

# XETRA holidays (Frankfurt Stock Exchange public holidays).
# Extend this set each December for the coming year.
_XETRA_HOLIDAYS = {
    # 2025
    (2025, 1, 1),   # New Year's Day
    (2025, 4, 18),  # Good Friday
    (2025, 4, 21),  # Easter Monday
    (2025, 5, 1),   # Labour Day
    (2025, 12, 24), # Christmas Eve (early close / closed)
    (2025, 12, 25), # Christmas Day
    (2025, 12, 26), # Boxing Day
    (2025, 12, 31), # New Year's Eve (early close / closed)
    # 2026
    (2026, 1, 1),   # New Year's Day
    (2026, 4, 3),   # Good Friday
    (2026, 4, 6),   # Easter Monday
    (2026, 5, 1),   # Labour Day
    (2026, 12, 24), # Christmas Eve
    (2026, 12, 25), # Christmas Day
    (2026, 12, 26), # Boxing Day
    (2026, 12, 31), # New Year's Eve
}


def _is_nyse_open() -> bool:
    """True if NYSE is currently open for regular trading (Mon–Fri 09:30–16:00 ET)."""
    now_utc = datetime.now(timezone.utc)
    now_et  = now_utc + timedelta(hours=_edt_offset(now_utc))

    if now_et.weekday() >= 5:
        return False

    today = (now_et.year, now_et.month, now_et.day)
    if today in _NYSE_HOLIDAYS:
        return False

    open_time  = now_et.replace(hour=9, minute=30, second=0, microsecond=0)
    close_hour = 13 if today in _NYSE_EARLY_CLOSE else 16
    close_time = now_et.replace(hour=close_hour, minute=0, second=0, microsecond=0)
    return open_time <= now_et < close_time


def _is_xetra_open() -> bool:
    """True if XETRA is currently open for regular trading (Mon–Fri 08:00–17:30 CET/CEST)."""
    now_utc  = datetime.now(timezone.utc)
    now_cet  = now_utc + timedelta(hours=_cet_offset(now_utc))

    if now_cet.weekday() >= 5:
        return False

    today = (now_cet.year, now_cet.month, now_cet.day)
    if today in _XETRA_HOLIDAYS:
        return False

    open_time  = now_cet.replace(hour=8,  minute=0,  second=0, microsecond=0)
    close_time = now_cet.replace(hour=17, minute=30, second=0, microsecond=0)
    return open_time <= now_cet < close_time


# Keep the old name as an alias so existing callers (tests, webhooks) don't break.
_is_market_open = _is_nyse_open


def _active_market() -> str | None:
    """Return which market is currently open: 'NYSE', 'XETRA', or None.
    NYSE takes priority during the overlap window (13:30–17:30 UTC in summer).
    """
    if _is_nyse_open():
        return "NYSE"
    if _is_xetra_open():
        return "XETRA"
    return None


def _auto_run_loop(interval_seconds: int):
    """Background thread: run the pipeline on a fixed schedule."""
    logger.info("[MAIN] Auto-run scheduler started — every %ds", interval_seconds)
    time.sleep(30)  # give Zeus time to fully initialise before first run
    while True:
        try:
            market = _active_market()
            if market is None:
                logger.info("[MAIN] Auto-run skipped — all markets closed")
            elif _zeus and _zeus.status.value != "halted":
                if _run_lock.acquire(blocking=False):
                    try:
                        logger.info("[MAIN] Auto-run triggered — market=%s", market)
                        runs = _zeus.run_once()
                        logger.info("[MAIN] Auto-run complete — %d signal(s) processed", len(runs))
                    finally:
                        _run_lock.release()
                else:
                    logger.info("[MAIN] Auto-run skipped — pipeline already running")
        except Exception as exc:
            logger.exception("[MAIN] Auto-run error: %s", exc)
        time.sleep(interval_seconds)


def run_webhook_server(host: str = "0.0.0.0", port: int = 8080):
    import threading
    from socketserver import ThreadingMixIn

    class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
        daemon_threads = True

    global _zeus
    _zeus = build_zeus()

    # Auto-run pipeline every RUN_INTERVAL seconds (default 15 min)
    interval = int(os.getenv("RUN_INTERVAL", "900"))
    t = threading.Thread(target=_auto_run_loop, args=(interval,), daemon=True)
    t.start()
    logger.info("[MAIN] Auto-scheduler: pipeline every %ds", interval)

    server = ThreadedHTTPServer((host, port), ZeusHandler)
    logger.info("[MAIN] ZEUS webhook server on %s:%d", host, port)
    logger.info("[MAIN] n8n → POST http://localhost:%d/run", port)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        logger.info("[MAIN] Shutting down.")
        _zeus.watchdog.stop()


# ---------------------------------------------------------------------------
# Standalone loop
# ---------------------------------------------------------------------------

def run_standalone(interval_seconds: int = 900):
    zeus = build_zeus()
    logger.info("[MAIN] Standalone mode — every %ds", interval_seconds)
    while True:
        try:
            runs = zeus.run_once()
            logger.info("[MAIN] Cycle complete — %d run(s).", len(runs))
        except Exception as exc:
            logger.exception("[MAIN] Cycle error: %s", exc)
        time.sleep(interval_seconds)


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    configure_logging()
    parser = argparse.ArgumentParser(description="ZEUS Trading Orchestrator")
    parser.add_argument("--standalone", action="store_true")
    parser.add_argument("--interval", type=int, default=900)
    parser.add_argument("--port",     type=int, default=8080)
    args = parser.parse_args()

    if args.standalone:
        run_standalone(interval_seconds=args.interval)
    else:
        run_webhook_server(port=args.port)