"""TGTC trading in-process service layer.

Provides:
  - TGTCAutoScheduler: asyncio-based intraday scheduler
  - tgtc_auto_scheduler: module-level singleton
  - Daemon launched as subprocess (python -m apps.tgtc_trader.daemon)
  - State persistence to .tgtc_auto_state.json
"""
from __future__ import annotations

import asyncio
import datetime as dt
import json
import logging
import os
import signal
import subprocess
import sys
import time
import traceback
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo

log = logging.getLogger(__name__)
_TZ_ET = ZoneInfo("America/New_York")
_DEFAULT_DB = "data/paper/tgtc.db"

_MARKET_OPEN = dt.time(9, 30)
_EOD_EXIT    = dt.time(15, 55)
_POST_CLOSE  = dt.time(16, 0)


# ── Saved-state helpers ───────────────────────────────────────────────────────

def _state_file_path(db_path: str) -> Path:
    return Path(db_path).parent / ".tgtc_auto_state.json"


def load_tgtc_saved_state(db_path: str) -> dict[str, Any] | None:
    p = _state_file_path(db_path)
    if not p.exists():
        return None
    try:
        return json.loads(p.read_text())
    except Exception:
        return None


def _save_state(db_path: str, running: bool, sessions: list[str], dry_run: bool) -> None:
    try:
        _state_file_path(db_path).write_text(json.dumps({
            "running": running,
            "sessions": sessions,
            "dry_run": dry_run,
            "db_path": db_path,
        }))
    except Exception:
        pass


def _clear_state(db_path: str) -> None:
    try:
        p = _state_file_path(db_path)
        if p.exists():
            p.unlink()
    except Exception:
        pass


# ── ORB-style daemon controller (subprocess) ──────────────────────────────────

_PROJECT_ROOT = Path(__file__).parent.parent.parent


class TGTCDaemonController:
    """Mirrors ORBDaemonController: manages TGTC daemon as a subprocess.

    Uses a PID file (.tgtc_scheduler.pid) as the authoritative liveness signal
    so uvicorn --reload hot-reloads cannot spawn duplicate daemons — a fresh
    controller instance checks the PID file before starting a new subprocess.

    Communication files:
      .tgtc_scheduler.pid     — daemon writes its PID on startup
      .tgtc_auto_state.json   — sessions / dry_run / db_path (written by controller)
      tgtc_scheduler.log      — human-readable log (written by daemon)
    """

    def __init__(self) -> None:
        self._process: subprocess.Popen | None = None  # type: ignore[type-arg]
        self._db_path: str = _DEFAULT_DB
        self._sessions: list[str] = []
        self._dry_run: bool = True

    def _pid_file(self) -> Path:
        return Path(self._db_path).parent / ".tgtc_scheduler.pid"

    def _log_file(self) -> Path:
        return Path(self._db_path).parent / "tgtc_scheduler.log"

    @property
    def running(self) -> bool:
        """True if the daemon process is alive (PID file + kill-0 check, zombie-safe)."""
        pid_file = self._pid_file()
        if not pid_file.exists():
            return False
        try:
            pid = int(pid_file.read_text().strip())
            os.kill(pid, 0)   # signal 0: probe without sending anything
            # kill-0 succeeds for zombie processes too — check ps to exclude zombies
            try:
                result = subprocess.run(
                    ["ps", "-p", str(pid), "-o", "stat="],
                    capture_output=True, text=True, timeout=2,
                )
                stat = result.stdout.strip()
                if stat.startswith("Z"):
                    # Zombie — treat as dead, clean up PID file
                    try:
                        pid_file.unlink()
                    except Exception:
                        pass
                    return False
            except Exception:
                pass  # if ps fails, trust kill-0 result
            return True
        except (ValueError, ProcessLookupError, PermissionError, OSError):
            # Process dead or PID stale — remove the file
            try:
                pid_file.unlink()
            except Exception:
                pass
            return False

    def start(
        self,
        sessions: list[str],
        db_path: str,
        dry_run: bool = True,
        start_now: bool = False,
        collect_duration_secs: int = 300,
        quick_end_after_mins: int = 60,
    ) -> None:
        self._db_path = db_path  # set FIRST so _pid_file() uses the right dir
        if self.running:
            raise RuntimeError("TGTCDaemonController already running")

        self._sessions = sessions
        self._dry_run = dry_run
        _save_state(db_path, True, sessions, dry_run)

        env = os.environ.copy()
        env["TGTC_TRADER_DB"] = db_path
        env["TGTC_DRY_RUN"] = "1" if dry_run else "0"
        env["TGTC_LOG_FILE"] = str(self._log_file())
        env["TGTC_START_NOW"] = "1" if start_now else "0"
        env["TGTC_COLLECT_DURATION_SECS"] = str(collect_duration_secs)
        env["TGTC_QUICK_END_AFTER_MINS"] = str(quick_end_after_mins)

        self._process = subprocess.Popen(
            [sys.executable, "-m", "apps.tgtc_trader.daemon"],
            env=env,
            cwd=str(_PROJECT_ROOT),
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
        log.info("TGTC daemon started (pid=%d, dry_run=%s)", self._process.pid, dry_run)

        # Wait up to 3 s for daemon to write its PID file
        for _ in range(30):
            time.sleep(0.1)
            if self._pid_file().exists():
                break

    def stop(self) -> None:
        """Send SIGTERM to daemon and clean up state files."""
        pid_file = self._pid_file()
        if pid_file.exists():
            try:
                pid = int(pid_file.read_text().strip())
                os.kill(pid, signal.SIGTERM)
            except Exception:
                pass
            try:
                pid_file.unlink()
            except Exception:
                pass
        elif self._process and self._process.poll() is None:
            self._process.terminate()
            try:
                self._process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                self._process.kill()
        self._process = None
        _clear_state(self._db_path)

    def shutdown(self) -> None:
        """Server shutdown — keep state so daemon can outlive the web server."""
        pass  # daemon runs in its own session, survives server restart

    def get_log(self, lines: int = 200) -> str:
        try:
            lf = self._log_file()
            if lf.exists():
                all_lines = lf.read_text(encoding="utf-8").splitlines()
                return "\n".join(all_lines[-lines:])
        except Exception:
            pass
        return ""

    def clear_log(self) -> None:
        try:
            lf = self._log_file()
            if lf.exists():
                lf.unlink()
        except Exception:
            pass

    def get_status(self) -> dict[str, Any]:
        pid_file = self._pid_file()
        pid: int | None = None
        if pid_file.exists():
            try:
                pid = int(pid_file.read_text().strip())
            except Exception:
                pass
        elif self._process:
            pid = self._process.pid
        return {
            "running": self.running,
            "pid": pid,
            "sessions": self._sessions,
            "dry_run": self._dry_run,
            "db_path": self._db_path,
        }


tgtc_auto_scheduler = TGTCDaemonController()


# ── Backtest task registry (in-process) ──────────────────────────────────────

import threading

_bt_tasks: dict[str, dict[str, Any]] = {}
_bt_lock = threading.Lock()
_BT_TASKS_DIR = Path("runs/.tgtc_bt_tasks")


def _save_bt_task(task_id: str, data: dict[str, Any]) -> None:
    try:
        _BT_TASKS_DIR.mkdir(parents=True, exist_ok=True)
        (_BT_TASKS_DIR / f"{task_id}.task.json").write_text(json.dumps(data, default=str))
    except Exception:
        pass


def _load_bt_tasks() -> dict[str, dict[str, Any]]:
    tasks = {}
    try:
        for p in _BT_TASKS_DIR.glob("*.task.json"):
            tid = p.stem.replace(".task", "")
            try:
                tasks[tid] = json.loads(p.read_text())
            except Exception:
                pass
    except Exception:
        pass
    return tasks


def get_bt_tasks() -> list[dict[str, Any]]:
    with _bt_lock:
        disk = _load_bt_tasks()
        merged = {**disk, **_bt_tasks}
    return sorted(merged.values(), key=lambda t: t.get("created_at", ""), reverse=True)


def get_bt_task(task_id: str) -> dict[str, Any] | None:
    with _bt_lock:
        if task_id in _bt_tasks:
            return _bt_tasks[task_id]
    p = _BT_TASKS_DIR / f"{task_id}.task.json"
    if p.exists():
        try:
            return json.loads(p.read_text())
        except Exception:
            pass
    return None


async def submit_backtest(
    date_str: str,
    config_path: str,
    universe: str | None = None,
    end_date_str: str | None = None,
) -> str:
    """Submit a TGTC backtest (single-day or multi-day). Returns task_id."""
    import uuid as _uuid

    task_id = _uuid.uuid4().hex[:8]
    now = dt.datetime.now(tz=dt.timezone.utc).isoformat()
    task: dict[str, Any] = {
        "task_id": task_id,
        "date": date_str,
        "end_date": end_date_str or date_str,
        "config_path": config_path,
        "universe": universe,
        "status": "pending",
        "created_at": now,
        "result": None,
        "error": None,
    }
    with _bt_lock:
        _bt_tasks[task_id] = task
    _save_bt_task(task_id, task)
    asyncio.create_task(_run_backtest_task(task_id, task))
    return task_id


async def _run_backtest_task(task_id: str, task: dict[str, Any]) -> None:
    with _bt_lock:
        _bt_tasks[task_id]["status"] = "running"

    try:
        multiday = task.get("end_date") and task["end_date"] != task["date"]
        runner = _run_multiday_backtest_sync if multiday else _run_backtest_sync
        result = await asyncio.to_thread(runner, task)
        with _bt_lock:
            _bt_tasks[task_id]["status"] = "completed"
            _bt_tasks[task_id]["result"] = result
        _save_bt_task(task_id, _bt_tasks[task_id])
    except Exception as exc:
        with _bt_lock:
            _bt_tasks[task_id]["status"] = "failed"
            _bt_tasks[task_id]["error"] = str(exc)
        _save_bt_task(task_id, _bt_tasks[task_id])
        log.error("TGTC backtest task %s failed: %s", task_id, exc, exc_info=True)


def _run_backtest_sync(task: dict[str, Any]) -> dict[str, Any]:
    """Synchronous backtest execution (runs in thread pool)."""
    import datetime as _dt

    date_str = task["date"]
    config_path = task["config_path"]
    universe_override = task.get("universe")

    from libs.tgtc.domain import load_tgtc_config
    from libs.tgtc.simulator import run_tgtc_simulation
    from libs.intraday.cache import IntradayCache, DailyBarCache
    from libs.intraday.features import enrich_daily_bars
    from apps.orb_trader.screener import load_universe
    from libs.tgtc.gainers_reconstruct import _bar_ts_naive_utc

    cfg = load_tgtc_config(config_path)
    if universe_override:
        cfg.universe = universe_override

    date = _dt.date.fromisoformat(date_str)
    tickers = load_universe(cfg.universe)

    intraday_cache = IntradayCache()
    daily_cache = DailyBarCache()

    # ── Load today's 5m bars ──────────────────────────────────────────────────
    bars_by_symbol: dict[str, list[dict]] = {}
    # Always include QQQ for regime filter, even if not in universe
    for sym in set(list(tickers) + ["QQQ"]):
        try:
            bars = intraday_cache.get(sym, date_str)
            if bars:
                bars_by_symbol[sym] = bars
        except Exception:
            pass

    # ── prev_close: from previous trading day's 5m bars ──────────────────────
    # Scan up to 5 calendar days back to find the last day with cached bars.
    prev_closes: dict[str, float] = {}
    close_16et_utc_hour = 20  # 16:00 EDT = 20:00 UTC
    for delta in range(1, 6):
        prev_date = (date - _dt.timedelta(days=delta)).isoformat()
        missing = [s for s in bars_by_symbol if s not in prev_closes]
        if not missing:
            break
        for sym in missing:
            try:
                prev_bars = intraday_cache.get(sym, prev_date)
                if not prev_bars:
                    continue
                # Use last bar at or before 16:00 ET as official close
                cutoff = _dt.datetime(
                    *[int(x) for x in prev_date.split("-")],
                    close_16et_utc_hour, 0
                )
                close_bar = None
                for b in prev_bars:
                    if _bar_ts_naive_utc(b) <= cutoff:
                        close_bar = b
                if close_bar:
                    prev_closes[sym] = float(close_bar["close"])
            except Exception:
                pass

    # ── Enrichment: ATR + avg_dollar_vol from daily cache ────────────────────
    enrichment: dict[str, dict] = {}
    start_daily = (date - _dt.timedelta(days=60)).isoformat()
    raw_daily: dict[str, list[dict]] = {}
    for sym in list(bars_by_symbol.keys()):
        try:
            rows, _ = daily_cache.get_with_tail(sym, start_daily, date_str)
            if rows:
                raw_daily[sym] = rows
        except Exception:
            pass

    if raw_daily:
        for sym, rows in raw_daily.items():
            # Sort and take only rows strictly before date_str (no lookahead)
            prior = sorted([r for r in rows if r.get("date", "") < date_str],
                           key=lambda r: r["date"])
            if not prior:
                continue
            # ATR-14 (simplified: avg of high-low over last 14 bars)
            last14 = prior[-14:]
            atr14 = (sum(float(b["high"]) - float(b["low"]) for b in last14) / len(last14)
                     if last14 else None)
            # avg dollar vol 30d
            last30 = prior[-30:]
            avg_dv = (sum(float(b["close"]) * float(b.get("volume", 0) or 0)
                          for b in last30) / len(last30)
                      if last30 else None)
            # prev_close = last available daily close
            daily_prev = float(prior[-1]["close"])
            enrichment[sym] = {
                "atr_14": atr14,
                "avg_dollar_vol_30d": avg_dv,
                "avg_dollar_vol_20d": avg_dv,
            }
            if daily_prev > 0 and sym not in prev_closes:
                prev_closes[sym] = daily_prev

    # ── QQQ pct change at 10:00 ET ───────────────────────────────────────────
    qqq_pct = None
    qqq_prev = prev_closes.get("QQQ")
    qqq_bars = bars_by_symbol.get("QQQ", [])
    if qqq_prev and qqq_prev > 0 and qqq_bars:
        cutoff = _dt.datetime(date.year, date.month, date.day, 14, 0)  # 10:00 EDT = 14:00 UTC
        bar10 = None
        for b in qqq_bars:
            if _bar_ts_naive_utc(b) <= cutoff:
                bar10 = b
        if bar10:
            qqq_pct = (float(bar10["close"]) - qqq_prev) / qqq_prev

    sim = run_tgtc_simulation(
        date=date,
        bars_by_symbol=bars_by_symbol,
        prev_closes=prev_closes,
        enrichment=enrichment,
        qqq_pct_change_at_10=qqq_pct,
        cfg=cfg,
    )

    trades_out = []
    for t in sim.trades:
        trades_out.append({
            "symbol": t.symbol,
            "entry_price": t.entry_price,
            "exit_price": t.exit_price,
            "stop_price": t.stop_price,
            "shares": t.shares,
            "entry_bar_idx": t.entry_bar_idx,
            "exit_bar_idx": t.exit_bar_idx,
            "exit_reason": t.exit_reason,
            "pnl": t.pnl,
            "r_multiple": t.r_multiple,
            "status": t.status,
            "tp_collision": getattr(t, "tp_collision", False),
            "side": getattr(t, "side", "long"),
            "entry_dt_utc": t.entry_dt_utc.isoformat() if t.entry_dt_utc else None,
            # Candidate metadata for segmentation analysis
            "score": getattr(t, "score", 0.0),
            "rank_persistence": getattr(t, "rank_persistence", 0.0),
            "rank_velocity": getattr(t, "rank_velocity", 0.0),
            "price_structure": getattr(t, "price_structure", 0.0),
            "volume_quality": getattr(t, "volume_quality", 0.0),
            "relative_strength": getattr(t, "relative_strength", 0.0),
            "pct_change_at_10": getattr(t, "pct_change_at_10", 0.0),
            "dollar_volume_20d": getattr(t, "dollar_volume_20d", 0.0),
        })

    return {
        "date": date_str,
        "n_trades": sim.n_trades,
        "total_pnl": round(sim.total_pnl, 2),
        "total_return_pct": round(sim.total_return_pct * 100, 2),
        "win_rate": round(sim.win_rate * 100, 1),
        "initial_equity": sim.initial_equity,
        "final_equity": sim.final_equity,
        "n_candidates": len(sim.candidates),
        "trades": trades_out,
        "equity_curve": sim.equity_curve,
        "candidates": [
            {k: v for k, v in c.items() if k not in ("bar_idx_10",)}
            for c in sim.candidates[:20]
        ],
    }


def _run_multiday_backtest_sync(task: dict[str, Any]) -> dict[str, Any]:
    """Loop over weekdays in [date, end_date], run single-day sim for each."""
    import datetime as _dt

    start = _dt.date.fromisoformat(task["date"])
    end = _dt.date.fromisoformat(task["end_date"])

    per_day: list[dict[str, Any]] = []
    all_trades: list[dict[str, Any]] = []
    equity = 10_000.0  # starting equity for cumulative curve

    # Read initial_equity from config if available
    try:
        from libs.tgtc.domain import load_tgtc_config
        cfg = load_tgtc_config(task["config_path"])
        equity = float(getattr(cfg, "initial_equity", 10_000.0))
    except Exception:
        pass

    initial_equity = equity
    equity_curve: list[dict[str, Any]] = []

    current = start
    while current <= end:
        if current.weekday() >= 5:  # skip weekends
            current += _dt.timedelta(days=1)
            continue

        day_task = {**task, "date": current.isoformat(), "end_date": current.isoformat()}
        try:
            day_result = _run_backtest_sync(day_task)
            daily_pnl = day_result.get("total_pnl", 0.0)
            equity += daily_pnl
            per_day.append({
                "date": current.isoformat(),
                "n_candidates": day_result.get("n_candidates", 0),
                "n_trades": day_result.get("n_trades", 0),
                "pnl": round(daily_pnl, 2),
                "return_pct": round(day_result.get("total_return_pct", 0.0), 2),
                "win_rate": round(day_result.get("win_rate", 0.0), 1),
                "equity": round(equity, 2),
            })
            for t in day_result.get("trades", []):
                all_trades.append({**t, "date": current.isoformat()})
            equity_curve.append({"date": current.isoformat(), "equity": round(equity, 2)})
        except Exception as exc:
            log.warning("TGTC multi-day: skipping %s — %s", current.isoformat(), exc)

        current += _dt.timedelta(days=1)

    n_trades = sum(d["n_trades"] for d in per_day)
    total_pnl = sum(d["pnl"] for d in per_day)
    wins = sum(1 for t in all_trades if t.get("pnl", 0) > 0)
    win_rate = round(wins / n_trades * 100, 1) if n_trades else 0.0

    return {
        "type": "multiday",
        "start_date": task["date"],
        "end_date": task["end_date"],
        "n_days": len(per_day),
        "n_trades": n_trades,
        "total_pnl": round(total_pnl, 2),
        "total_return_pct": round((equity - initial_equity) / initial_equity * 100, 2),
        "win_rate": win_rate,
        "initial_equity": initial_equity,
        "final_equity": round(equity, 2),
        "per_day": per_day,
        "trades": all_trades,
        "equity_curve": equity_curve,
    }