#!/usr/bin/env python3
"""AlphaBot V2: stdlib CLI, streamed prices, SQLite macro/events and durable risk state.
Live mode speaks the documented AlphaBot gateway contract, never a broker directly.
"""
import argparse
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
import fcntl
import getpass
import hashlib
import http.cookiejar
import json
import os
from pathlib import Path
import platform
import resource
import re
import signal
import sqlite3
import ssl
import sys
import time
import threading
import urllib.error
import urllib.parse
import urllib.request
import uuid
from host_strategy import capital_floor, decide

D = Decimal
STOP = False


class Fault(Exception):
    pass


def number(value):
    try:
        result = D(str(value))
        if not result.is_finite():
            raise ValueError
        return result
    except (InvalidOperation, ValueError):
        raise Fault('invalid finite numeric value') from None


def stamp(value):
    text = str(value)
    if len(text) == 10:
        text += 'T00:00:00+00:00'
    try:
        result = datetime.fromisoformat(text.replace('Z', '+00:00'))
        if result.tzinfo is None:
            raise ValueError
        return result.astimezone(timezone.utc)
    except ValueError:
        raise Fault('timestamp requires ISO date or timezone-aware ISO timestamp') from None


def emit(event, **values):
    print(json.dumps({'event': event, **values}, default=str, allow_nan=False), flush=True)


def records(path):
    with path.open(encoding='utf-8') as stream:
        line_no = 0
        while True:
            line = stream.readline(4 * 1024 * 1024 + 1)
            if not line:
                break
            line_no += 1
            if len(line) > 4 * 1024 * 1024:
                raise Fault('JSONL record exceeds 4 MiB')
            if not line.strip():
                continue
            try:
                value = json.loads(line)
                if not isinstance(value, dict):
                    raise ValueError
            except (ValueError, TypeError):
                raise Fault(f'malformed JSONL at line {line_no}') from None
            yield value


class Store:
    def __init__(self, root, mode):
        self.lock = (root / f'{mode}.lock').open('a')
        try:
            fcntl.flock(self.lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            self.lock.close()
            raise Fault('another runner owns this project/mode') from None
        self.db = sqlite3.connect(root / f'{mode}.sqlite3')
        self.db.executescript('''PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;
        CREATE TABLE IF NOT EXISTS state(k TEXT PRIMARY KEY,v TEXT NOT NULL);
        CREATE TABLE IF NOT EXISTS macro(obs TEXT, available TEXT, value TEXT,
          PRIMARY KEY(obs,available));
        CREATE INDEX IF NOT EXISTS macro_available ON macro(available);
        CREATE TABLE IF NOT EXISTS geo(at TEXT PRIMARY KEY,tone TEXT);
        CREATE TABLE IF NOT EXISTS audit(id INTEGER PRIMARY KEY,body TEXT NOT NULL);''')

    def get(self, key, default=None):
        row = self.db.execute('SELECT v FROM state WHERE k=?', (key,)).fetchone()
        return json.loads(row[0]) if row else default

    def put(self, **values):
        with self.db:
            for key, value in values.items():
                self.db.execute('INSERT OR REPLACE INTO state VALUES(?,?)',
                                (key, json.dumps(value, default=str)))

    def audit(self, event, **values):
        body = {'event': event, **values}
        with self.db:
            self.db.execute('INSERT INTO audit(body) VALUES(?)',
                            (json.dumps(body, default=str),))
        emit(event, **values)

    def ingest(self, root):
        # Rebuild indexes transactionally; bounded memory regardless of file size.
        with self.db:
            self.db.execute('DELETE FROM macro')
            self.db.execute('DELETE FROM geo')
            for row in records(root / 'macro.jsonl'):
                if row['series'] != 'INDPRO':
                    continue
                obs = stamp(row['observation_date'])
                supplied = row.get('available_at', row.get('publication_date'))
                available = obs if obs.year < 2000 or not supplied else stamp(supplied)
                if available < obs:
                    raise Fault('macro availability precedes observation')
                self.db.execute('INSERT OR REPLACE INTO macro VALUES(?,?,?)',
                                (obs.isoformat(), available.isoformat(), str(number(row['value']))))
            path = root / 'geopolitical.jsonl'
            if not path.is_file():
                raise Fault('geopolitical.jsonl required; empty file explicitly means no events')
            for row in records(path):
                at = stamp(row['at']).isoformat()
                # Preserve simultaneous events by conservatively retaining worst tone.
                old = self.db.execute('SELECT tone FROM geo WHERE at=?', (at,)).fetchone()
                tone = number(row['tone'])
                if old:
                    tone = min(tone, number(old[0]))
                self.db.execute('INSERT OR REPLACE INTO geo VALUES(?,?)', (at, str(tone)))

    def indicators(self, at):
        # Select latest visible vintage per month, then demand four consecutive months.
        months = {}
        for obs, available, value in self.db.execute(
                'SELECT obs,available,value FROM macro WHERE obs<=? AND available<=? '
                'ORDER BY obs DESC,available DESC', (at.isoformat(), at.isoformat())):
            month = obs[:7]
            if month not in months:
                months[month] = number(value)
            if len(months) == 4:
                break
        keys = sorted(months)
        indices = [int(k[:4]) * 12 + int(k[5:7]) for k in keys]
        values = [months[k] for k in keys] if len(indices) == 4 and indices[-1] - indices[0] == 3 else []
        # Never expose future events merely because their calendar month has begun.
        tones = (number(r[0]) for r in self.db.execute(
            'SELECT tone FROM geo WHERE at>=? AND at<=?',
            (at.replace(day=1, hour=0, minute=0, second=0, microsecond=0).isoformat(), at.isoformat())))
        count = self.db.execute('SELECT count(*) FROM geo WHERE substr(at,1,7)=?',
                                (at.strftime('%Y-%m'),)).fetchone()[0]
        return values, min(tones, default=D('0') if count == 0 else None), bool(count == 0), keys[-1] if keys else ''


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *args, **kwargs):
        return None


class AuthenticationFailed(Fault):
    def __init__(self):
        super().__init__('AUTHENTICATION_FAILED')


class Gateway:
    def __init__(self, cfg):
        self.broker = cfg['broker']
        self.account_id = cfg['account_id']
        self.deadline = 0.0
        self.auth_lock = threading.Lock()
        self.revoked = threading.Event()
        self.closed = threading.Event()
        self.revoked.set()
        self.cookies = http.cookiejar.CookieJar()
        if self.broker not in ('alpaca', 'kraken', 'bitstamp'):
            raise Fault('unsupported broker')
        self.base = os.environ.get('LIVE_AUTH_GATEWAY_URL', '')
        if self.base and '://' not in self.base:
            self.base = 'https://' + self.base
        parsed = urllib.parse.urlsplit(self.base)
        if (parsed.scheme != 'https' or not parsed.hostname or parsed.username or
                parsed.password or parsed.query or parsed.fragment or parsed.path not in ('', '/')):
            raise AuthenticationFailed()
        if parsed.hostname != cfg.get('gateway_hostname'):
            raise AuthenticationFailed()
        self.hostname = parsed.hostname
        self.base = self.base.rstrip('/')
        self.http = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect(),
            urllib.request.HTTPCookieProcessor(self.cookies),
            urllib.request.HTTPSHandler(context=ssl.create_default_context()))
        if platform.system() != 'Darwin' or not sys.stdin.isatty():
            raise AuthenticationFailed()
        resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
        key = os.environ.get('LIVE_BROKER_API_KEY')
        secret = os.environ.get('LIVE_BROKER_SECRET_KEY')
        try:
            if not key or not secret:
                raise AuthenticationFailed()
            self.authenticate()
            # Separate authenticated bootstrap, never extra fields in the gate payload.
            result = self.call('/v1/broker-session', {'broker': self.broker,
                'account_id': self.account_id, 'api_key': key, 'secret_key': secret})
            if result.get('bound') is not True or result.get('account_id') != self.account_id:
                raise AuthenticationFailed()
        except BaseException:
            self.cookies.clear()
            self.revoked.set()
            raise
        finally:
            key = secret = None
            os.environ.pop('LIVE_BROKER_API_KEY', None)
            os.environ.pop('LIVE_BROKER_SECRET_KEY', None)
        self.monitor = threading.Thread(target=self.heartbeat, name='auth-expiry', daemon=True)
        self.monitor.start()
        # Reference disposal is best effort; immutable strings cannot be securely wiped.

    def heartbeat(self):
        while not self.closed.wait(0.1):
            with self.auth_lock:
                if time.monotonic() >= self.deadline:
                    self.revoked.set()

    def authenticate(self):
        self.revoked.set()
        self.cookies.clear()
        password = challenge = None
        try:
            password = getpass.getpass('AlphaBot gateway password: ')
            challenge = hashlib.sha256(password.encode('utf-8')).hexdigest()
            password = None
            started = time.monotonic()
            result = self.call('/api/v1/auth/gate', {'gate_challenge': challenge}, auth=False)
            if (result.get('status') != 'AUTHENTICATED' or
                    type(result.get('session_token_expires_sec')) is not int or
                    result['session_token_expires_sec'] != 3600):
                raise AuthenticationFailed()
            # The specified JSON contains no token: require an authenticated cookie.
            usable = any(c.secure and c.domain.lstrip('.') == self.hostname and
                         c.path == '/' and c.has_nonstandard_attr('HttpOnly') and
                         (c.expires is None or c.expires > time.time()) for c in self.cookies)
            if not usable:
                raise AuthenticationFailed()
            with self.auth_lock:
                self.deadline = started + 3600  # Network time never extends the session.
                self.revoked.clear()
        except BaseException:
            self.cookies.clear()
            self.revoked.set()
            raise AuthenticationFailed() from None
        finally:
            password = challenge = None

    def ensure_auth(self):
        # Called in the main thread; the heartbeat never reads the terminal.
        if self.revoked.is_set() or time.monotonic() >= self.deadline:
            self.authenticate()
        if self.revoked.is_set() or time.monotonic() >= self.deadline:
            raise AuthenticationFailed()

    def call(self, path, body, auth=True):
        if auth:
            self.ensure_auth()
        request = urllib.request.Request(self.base + path,
            data=json.dumps(body, default=str).encode(),
            headers={'Content-Type': 'application/json', 'Accept': 'application/json'}, method='POST')
        if auth and (self.revoked.is_set() or time.monotonic() >= self.deadline):
            raise AuthenticationFailed()
        with self.http.open(request, timeout=15) as response:
            raw = response.read(2 * 1024 * 1024 + 1)
            if len(raw) > 2 * 1024 * 1024:
                raise Fault('oversized gateway response')
            result = json.loads(raw)
            if not isinstance(result, dict):
                raise Fault('invalid gateway response')
            return result

    def snapshot(self, max_age):
        row = self.call('/v1/snapshot', {'broker': self.broker})
        age = time.time() - stamp(row['at']).timestamp()
        if age < -5 or age > max_age:
            raise Fault('stale account snapshot')
        if row.get('account_id') != self.account_id or type(row.get('flat')) is not bool:
            raise Fault('account identity/position status mismatch')
        if row.get('currency') != 'USD' or row.get('long_only') is not True:
            raise Fault('USD long-only account required')
        for field in ('cash', 'equity', 'asset_cost', 'exit_fees'):
            row[field] = number(row[field])
            if row[field] < 0:
                raise Fault('negative account field')
        if type(row.get('open_orders')) is not int or row['open_orders'] < 0:
            raise Fault('invalid open order count')
        return row

    def close(self):
        self.revoked.set()
        self.closed.set()
        monitor = getattr(self, 'monitor', None)
        if monitor:
            monitor.join(timeout=1)
        self.cookies.clear()


def route(broker, symbol, quantity, client_id):
    """Native order mapping; the gateway signs, rounds, validates and reconciles."""
    if broker not in ('alpaca', 'kraken', 'bitstamp'):
        raise Fault('unsupported broker')
    if broker == 'alpaca':
        if not re.fullmatch(r'[A-Z][A-Z0-9.\-]{0,14}', symbol):
            raise Fault('Alpaca route is reserved for equities')
        return {'method': 'POST', 'path': '/v2/orders', 'body': {
            'symbol': symbol, 'qty': str(quantity), 'side': 'buy', 'type': 'market',
            'time_in_force': 'day', 'client_order_id': client_id}}
    if not re.fullmatch(r'[A-Z0-9]{2,12}/USD', symbol):
        raise Fault('crypto route requires BASE/USD symbol')
    if broker == 'kraken':
        return {'method': 'POST', 'path': '/0/private/AddOrder', 'body': {
            'pair': symbol, 'type': 'buy', 'ordertype': 'market',
            'volume': str(quantity), 'cl_ord_id': client_id}}
    return {'method': 'POST', 'path': '/api/v2/buy/market/' + symbol.replace('/', '').lower() + '/',
            'body': {'amount': str(quantity), 'client_order_id': client_id}}


class Engine:
    def __init__(self, store, cfg, live=False):
        self.s, self.cfg = store, cfg
        self.live = live
        self.g = None
        self.capital = number(cfg['starting_capital'])
        self.fee = number(cfg.get('fee_rate', '0.001'))
        self.slip = number(cfg.get('slippage_rate', '0.001'))
        self.symbol = cfg['index_asset']
        if self.capital <= 0 or not 0 <= self.fee < 1 or not 0 <= self.slip < 1:
            raise Fault('invalid capital/fee/slippage configuration')
        fingerprint = hashlib.sha256(json.dumps(cfg, sort_keys=True).encode()).hexdigest()
        if store.get('config_hash', fingerprint) != fingerprint:
            raise Fault('configuration differs from durable run state; use a new project')
        store.put(config_hash=fingerprint)
        self.cash = number(store.get('cash', str(self.capital)))
        self.qty = number(store.get('qty', '0'))
        self.cost = number(store.get('cost', '0'))
        self.peak = number(store.get('peak', str(self.capital)))
        self.halted = store.get('halted', False)
        self.last_entry = store.get('last_entry', '')
        self.last_at = store.get('last_at', '')
        if live:
            self.g = Gateway(cfg)

    def save(self):
        self.s.put(cash=self.cash, qty=self.qty, cost=self.cost, peak=self.peak,
                   halted=self.halted, last_entry=self.last_entry, last_at=self.last_at)

    def account(self, price):
        if self.g:
            a = self.g.snapshot(self.cfg.get('max_snapshot_age_seconds', 15))
            self.cash, self.cost = a['cash'], a['asset_cost']
            equity, exit_fees = a['equity'], a['exit_fees']
        else:
            equity = self.cash + self.qty * price
            exit_fees = self.qty * price * (self.fee + self.slip)
            a = {'equity': equity, 'open_orders': 0}
        self.peak = max(self.peak, equity)
        self.save()  # Persist milestone before strategy/order dispatch.
        floor = capital_floor(self.capital, self.cost, self.peak)
        return equity, exit_fees, floor, a

    def halt(self, reason, price):
        self.halted = True
        self.save()  # Latch halt before contacting any order endpoint.
        self.s.audit('SAFE_HALT', reason=reason)
        if self.g:
            # Server must cancel pending orders, reconcile, then close all long positions.
            identity = self.s.get('liquidation_id') or str(uuid.uuid4())
            self.s.put(liquidation_id=identity)
            result = self.g.call('/v1/liquidate', {
                'broker': self.cfg['broker'], 'idempotency_key': identity,
                'reason': reason, 'cancel_open_orders': True})
            verified = self.g.snapshot(self.cfg.get('max_snapshot_age_seconds', 15))
            if result.get('flat') is not True or verified.get('flat') is not True or verified['open_orders']:
                raise Fault('liquidation pending/unconfirmed; strategy remains halted')
        elif self.qty and price > 0:
            proceeds = self.qty * price * (1 - self.slip) * (1 - self.fee)
            self.cash += proceeds
            self.qty = self.cost = D('0')
        self.save()
        if not self.g and self.qty:
            raise Fault('missing liquidation price; position remains halted')
        self.s.audit('LIQUIDATION_CONFIRMED', cash=self.cash)

    def tick(self, at, price, values, tone, neutral, month):
        equity, fees, floor, account = self.account(price)
        self.s.audit('TELEMETRY', at=at, price=price, equity=equity, cash=self.cash,
            asset_cost=self.cost, peak=self.peak, floor=floor, modeled_exit_fees=fees,
            indpro=values, tone=tone, empty_event_month=neutral)
        if self.halted or self.s.get('pending'):
            self.halt('durable_halt_or_unreconciled_order', price)
            return
        if equity - fees < floor or (tone is not None and tone < D('-5')):
            self.halt('floor_threat' if equity - fees < floor else 'geopolitical_tone', price)
            return
        if self.g:
            self.g.ensure_auth()  # No strategy callback after expiry without a fresh challenge.
        action = decide({'cash': self.cash, 'indpro': values, 'tone': tone,
                         'month': month, 'last_entry_month': self.last_entry})
        if action['action'] != 'buy' or account['open_orders']:
            return
        # Re-evaluate account immediately before dispatch; gateway must enforce atomically too.
        equity, fees, floor, account = self.account(price)
        if equity - fees < floor:
            self.halt('pre_order_floor_threat', price)
            return
        if account['open_orders']:
            return
        budget = min(number(action['budget']), self.cash * D('.20'))
        fill = price * (1 + self.slip)
        qty = (budget / (fill * (1 + self.fee))).quantize(D('.00000001'), rounding='ROUND_DOWN')
        spend = qty * fill * (1 + self.fee)
        if qty <= 0:
            return
        projected = equity - (spend - qty * price)
        projected_floor = capital_floor(self.capital, self.cost + spend, max(self.peak, projected))
        if projected - fees - qty * price * (self.fee + self.slip) < projected_floor:
            self.s.audit('BUY_BLOCKED', reason='projected_floor')
            return
        identity = str(uuid.uuid4())
        self.s.put(pending=identity)
        if self.g:
            result = self.g.call('/v1/order', {
                'broker': self.cfg['broker'], 'idempotency_key': identity,
                'native_order': route(self.cfg['broker'], self.symbol, qty, identity),
                'max_total_debit': str(budget), 'starting_capital': str(self.capital),
                'peak_equity': str(self.peak), 'floor': str(floor),
                'require_atomic_floor_check': True})
            if result.get('status') != 'filled' or result.get('idempotency_key') != identity:
                raise Fault('order not fully reconciled; halt required')
            # Independently fetch authoritative totals; never model live fills as completed.
            self.account(price)
        else:
            self.cash -= spend
            self.qty += qty
            self.cost += spend
        self.last_entry = month
        self.save()
        self.s.put(pending=None)
        self.s.audit('BUY_CONFIRMED', id=identity, symbol=self.symbol, modeled_quantity=qty)


def run(args):
    root = Path(args.project).expanduser().resolve(strict=True)
    cfg = json.loads((root / 'config.json').read_text())
    until = stamp(args.until)
    if not 1 <= args.start_year <= until.year:
        raise Fault('invalid start year/until range')
    live = args.mode == 'live'
    if live and not args.continuous:
        raise Fault('live mode requires --continuous')
    store = Store(root, args.mode)
    engine = None
    price = None
    try:
        store.ingest(root)
        engine = Engine(store, cfg, live)
        store.audit('START', mode=args.mode,
                    macro_alignment='pre2000_observation_date; later_available_at_or_observation_assumed',
                    strategy='monthly_INDPRO_vs_previous_3_months')
        if live:
            equity, fees, floor, _ = engine.account(D('0'))
            if engine.halted or store.get('pending') or equity - fees < floor:
                engine.halt('startup_risk_intercept', D('0'))
                return 2
        previous = None
        with (root / 'prices.jsonl').open(encoding='utf-8') as stream:
            while not STOP:
                if live and datetime.now(timezone.utc) >= until:
                    break
                position = stream.tell()
                line = stream.readline(4 * 1024 * 1024 + 1)
                if len(line) > 4 * 1024 * 1024:
                    raise Fault('price record exceeds 4 MiB')
                if not line or not line.endswith('\n'):
                    stream.seek(position)
                    if args.continuous and datetime.now(timezone.utc) < until:
                        if live:
                            equity, fees, floor, _ = engine.account(price or D('0'))
                            store.ingest(root)
                            _, tone, _, _ = store.indicators(datetime.now(timezone.utc))
                            if (engine.halted or store.get('pending') or equity - fees < floor
                                    or (tone is not None and tone < D('-5'))):
                                engine.halt('idle_risk_intercept', price or D('0'))
                                break
                        time.sleep(1)
                        continue
                    if line:
                        raise Fault('incomplete final price record')
                    break
                row = json.loads(line)
                at = stamp(row['at'])
                if previous is not None and at <= previous:
                    raise Fault('prices must have strictly increasing timestamps')
                previous = at
                if at.year < args.start_year or (engine.last_at and at <= stamp(engine.last_at)):
                    continue
                if at > until:
                    break
                if live:
                    age = time.time() - at.timestamp()
                    if age > cfg.get('max_price_age_seconds', 30):
                        continue  # Historical replay must never dispatch live orders.
                    if age < -5:
                        raise Fault('future live price timestamp')
                    store.ingest(root)  # Refresh local macro/event files at each live tick.
                price = number(row['quotes'][engine.symbol])
                if price <= 0:
                    raise Fault('price must be positive')
                values, tone, neutral, month = store.indicators(at)
                engine.tick(at, price, values, tone, neutral, month)
                engine.last_at = at.isoformat()
                engine.save()
                if engine.halted:
                    break
        if live and price is None:
            engine.halt('no_fresh_price_before_stop', D('0'))
        elif live and not engine.halted:
            engine.halt('runner_stopped_or_until_reached', price)
        store.audit('COMPLETE', halted=engine.halted, last_at=engine.last_at)
        return 2 if engine.halted else 0
    except BaseException as exc:
        # Avoid exception messages/tracebacks that could contain HTTP headers or secrets.
        store.audit('AUTHENTICATION_FAILED' if isinstance(exc, AuthenticationFailed) else 'ERROR',
                    kind=type(exc).__name__)
        if engine is None and live:
            store.put(halted=True)
        if engine is not None and isinstance(exc, AuthenticationFailed):
            engine.halted = True
            engine.save()
            store.audit('LIQUIDATION_UNCONFIRMED', intervention_required=True)
        elif engine is not None:
            try:
                engine.halt('execution_error', price or D('0'))
            except BaseException:
                store.audit('LIQUIDATION_UNCONFIRMED', intervention_required=True)
        return 2
    finally:
        if engine and engine.g:
            engine.g.close()
        store.db.close()
        store.lock.close()


def stop(signum, frame):
    global STOP
    STOP = True


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--project', required=True)
    parser.add_argument('--start-year', required=True, type=int)
    parser.add_argument('--until', required=True)
    parser.add_argument('--continuous', action='store_true')
    parser.add_argument('--mode', choices=('backtest', 'live'), default='backtest')
    args = parser.parse_args()
    signal.signal(signal.SIGTERM, stop)
    signal.signal(signal.SIGINT, stop)
    try:
        return run(args)
    except (OSError, ValueError, KeyError, Fault):
        emit('STARTUP_REJECTED', reason='invalid project, configuration, or input')
        return 2


if __name__ == '__main__':
    raise SystemExit(main())