#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
"""AlphaBot V2 native broker execution candidate, 1.0.0-rc1.

Real Alpaca REST execution + authenticated data/order WebSockets; Kraken signed
REST + authenticated WebSocket v2 execution. One dedicated USD long-only spot
account per project. Independent OS thread owns the 1 Hz risk loop. No backtest
or demo execution path. `check` never submits orders. `run --live` does.
"""
import argparse
import asyncio
import base64
import contextlib
from datetime import datetime, timezone, timedelta
from decimal import Decimal, ROUND_DOWN, getcontext
import fcntl
import hashlib
import hmac
import json
import os
from pathlib import Path
import re
import signal
import sqlite3
import threading
import time
import urllib.parse
import uuid

import aiohttp
from host_strategy import capital_floor, decide, decimal, risk_multiplier, target_deviation

VERSION = '1.0.0-rc1'
D = Decimal
getcontext().prec = 50
ZERO = D('0')
TERMINAL = {'filled', 'canceled', 'cancelled', 'expired', 'rejected', 'closed'}


class Fault(Exception):
    """A message safe to show: never include upstream bodies or credentials."""


def now():
    return datetime.now(timezone.utc).isoformat()


def timestamp(value):
    if len(value) == 10:
        value += 'T00:00:00+00:00'
    result = datetime.fromisoformat(value.replace('Z', '+00:00'))
    if result.tzinfo is None:
        raise Fault('timezone_required')
    return result.astimezone(timezone.utc)


def fixed(value):
    if isinstance(value, D):
        if not value.is_finite():
            raise ValueError('nonfinite_decimal')
        return format(value, 'f')
    if isinstance(value, dict):
        return {str(k): fixed(v) for k, v in value.items()}
    if isinstance(value, (tuple, list)):
        return [fixed(v) for v in value]
    if isinstance(value, float):
        raise ValueError('binary_float_in_accounting')
    return value


def encode(value):
    return json.dumps(fixed(value), sort_keys=True, separators=(',', ':'), allow_nan=False)


def wire(value):
    """Kraken requires numeric JSON tokens, without a binary-float conversion."""
    if isinstance(value, D):
        return format(decimal(value), 'f')
    if isinstance(value, dict):
        return '{' + ','.join(json.dumps(k) + ':' + wire(v) for k, v in value.items()) + '}'
    if isinstance(value, list):
        return '[' + ','.join(map(wire, value)) + ']'
    return json.dumps(value, allow_nan=False)


def decode(value):
    def invalid(_):
        raise Fault('nonfinite_json')
    return json.loads(value, parse_float=D, parse_constant=invalid)


def log(event, **data):
    print(encode({'at': now(), 'event': event, **data}), flush=True)


def configuration(root):
    cfg = decode((root / 'config.json').read_text())
    if cfg.get('broker') not in ('alpaca', 'kraken'):
        raise Fault('broker_must_be_alpaca_or_kraken')
    symbol = cfg.get('symbol', '')
    if cfg['broker'] == 'alpaca' and not re.fullmatch(r'[A-Z][A-Z0-9.\-]{0,14}', symbol):
        raise Fault('invalid_equity_symbol')
    if cfg['broker'] == 'kraken' and not re.fullmatch(r'[A-Z0-9]+/USD', symbol):
        raise Fault('kraken_requires_BASE_USD')
    defaults = {'starting_capital': '500', 'lesson_risk_decay': '0.95',
        'execution_target_multiplier': '1', 'fee_reserve_rate': '0.005',
        'slippage_reserve_rate': '0.01', 'max_order_usd': '100',
        'market_data_max_age_seconds': 120, 'alpaca_feed': 'iex',
        'dedicated_account': True}
    cfg = {**defaults, **cfg}
    for key in ('starting_capital', 'max_order_usd', 'execution_target_multiplier'):
        if decimal(cfg[key]) <= 0:
            raise Fault('positive_configuration_required_' + key)
    for key in ('fee_reserve_rate', 'slippage_reserve_rate'):
        if not 0 <= decimal(cfg[key]) < D('0.25'):
            raise Fault('invalid_reserve_' + key)
    if cfg['dedicated_account'] is not True:
        raise Fault('dedicated_account_required')
    if cfg['alpaca_feed'] not in ('iex', 'sip'):
        raise Fault('invalid_alpaca_feed')
    if type(cfg['market_data_max_age_seconds']) is not int or not 60 <= cfg['market_data_max_age_seconds'] <= 300:
        raise Fault('market_data_age_must_be_60_to_300_seconds')
    risk_multiplier(0, cfg['lesson_risk_decay'])
    return cfg


class Journal:
    """FULL-sync WAL transactions; shared across risk/strategy via an RLock.

    Observations are an internal management ledger, not a GAAP/tax book. The
    append-only hash chain detects edits, not a privileged attack rewriting it all.
    Pending intents are committed BEFORE network submission; never blindly replay.
    """
    def __init__(self, root, cfg, key_fingerprint):
        self.mutex = threading.RLock()
        self.halted = threading.Event()
        self.kraken_gate = threading.Lock()
        self.order_gate = threading.Lock()
        self.root, self.cfg = root, cfg
        self.lock_file = (root / 'native-live.lock').open('a')
        try:
            fcntl.flock(self.lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            self.lock_file.close()
            raise Fault('project_already_running') from None
        try:
            self.db = sqlite3.connect(root / 'native-live.sqlite3', check_same_thread=False, timeout=5)
            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 events(id INTEGER PRIMARY KEY,at TEXT NOT NULL,
              kind TEXT NOT NULL,body TEXT NOT NULL,parent TEXT NOT NULL,hash TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS intents(client_id TEXT PRIMARY KEY,body TEXT NOT NULL,
              status TEXT NOT NULL,broker_id TEXT);
            CREATE TABLE IF NOT EXISTS receipts(broker_id TEXT PRIMARY KEY,body TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS lessons(id INTEGER PRIMARY KEY,event_id INTEGER,
              body TEXT NOT NULL);''')
            os.chmod(root / 'native-live.sqlite3', 0o600)
            self.verify()
            binding = hashlib.sha256(encode({'config': cfg, 'key_fingerprint': key_fingerprint}).encode()).hexdigest()
            with self.mutex, self.db:
                old = self.get('binding')
                if old and old != binding:
                    raise Fault('durable_account_or_config_binding_mismatch')
                self.set('binding', binding)
                if self.get('halt'):
                    self.halted.set()
                if self.db.execute("SELECT count(*) FROM intents WHERE status='pending'").fetchone()[0]:
                    self.halt('restart_with_unresolved_order_intent')
        except BaseException:
            with contextlib.suppress(Exception):
                self.db.close()
            self.lock_file.close()
            raise

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

    def set(self, key, value):
        self.db.execute('INSERT OR REPLACE INTO state VALUES(?,?)', (key, encode(value)))

    def _append(self, kind, body):
        at = now()
        parent = self.db.execute('SELECT hash FROM events ORDER BY id DESC LIMIT 1').fetchone()
        parent = parent[0] if parent else '0' * 64
        body = encode(body)
        digest = hashlib.sha256(encode([at, kind, body, parent]).encode()).hexdigest()
        cursor = self.db.execute('INSERT INTO events(at,kind,body,parent,hash) VALUES(?,?,?,?,?)',
                                 (at, kind, body, parent, digest))
        return cursor.lastrowid

    def append(self, kind, body):
        with self.mutex, self.db:
            return self._append(kind, body)

    def verify(self):
        parent = '0' * 64
        with self.mutex:
            for at, kind, body, previous, digest in self.db.execute(
                    'SELECT at,kind,body,parent,hash FROM events ORDER BY id'):
                if previous != parent or hashlib.sha256(encode([at, kind, body, previous]).encode()).hexdigest() != digest:
                    raise Fault('journal_integrity_failure')
                parent = digest
            if self.db.execute('PRAGMA quick_check').fetchone()[0] != 'ok':
                raise Fault('sqlite_integrity_failure')

    def lesson_count(self):
        with self.mutex:
            return self.db.execute('SELECT count(*) FROM lessons').fetchone()[0]

    def _lesson(self, path, details):
        count = self.lesson_count()
        body = {'failure_path': path, 'details': details, 'cumulative_count': count + 1,
                'multiplier_before': risk_multiplier(count, self.cfg['lesson_risk_decay']),
                'multiplier_after': risk_multiplier(count + 1, self.cfg['lesson_risk_decay'])}
        event_id = self._append('LESSON', body)
        self.db.execute('INSERT INTO lessons(event_id,body) VALUES(?,?)', (event_id, encode(body)))

    def halt(self, reason):
        # Admission stops even if a database commit fails; I/O failure must not allow a buy.
        self.halted.set()
        with self.mutex, self.db:
            if not self.get('halt'):
                self.set('halt', {'at': now(), 'reason': reason})
                self._append('HALT', {'reason': reason})
        log('HALT', reason=reason)

    def intent(self, side, quantity, reference, month='', emergency=False):
        with self.mutex, self.db:
            if self.halted.is_set() and not emergency:
                raise Fault('durable_halt_blocks_entry')
            cid = str(uuid.uuid4())
            body = {'side': side, 'quantity': quantity, 'reference': reference,
                    'month': month, 'symbol': self.cfg['symbol'], 'emergency': emergency}
            self.db.execute('INSERT INTO intents VALUES(?,?,?,NULL)', (cid, encode(body), 'pending'))
            self._append('ORDER_INTENT', {'client_id': cid, **body})
            self.set('execution_generation', int(self.get('execution_generation', 0)) + 1)
            if side == 'buy':
                self.set('last_entry_month', month)  # A partial/ambiguous entry consumes its month.
            return cid

    def receipt(self, cid, result):
        """One terminal aggregate receipt per broker order; no partial double counting."""
        with self.mutex, self.db:
            row = self.db.execute('SELECT body FROM intents WHERE client_id=?', (cid,)).fetchone()
            if not row:
                raise Fault('receipt_without_intent')
            intent = decode(row[0])
            bid = result['broker_id']
            prior = self.db.execute('SELECT body FROM receipts WHERE broker_id=?', (bid,)).fetchone()
            if prior:
                if decode(prior[0]) != fixed(result):
                    raise Fault('conflicting_terminal_receipt')
                return
            self.db.execute('INSERT INTO receipts VALUES(?,?)', (bid, encode(result)))
            self.db.execute('UPDATE intents SET status=?,broker_id=? WHERE client_id=?',
                            (result['status'], bid, cid))
            qty, price = decimal(result['quantity']), decimal(result['price'])
            if qty > 0:
                held, cost = decimal(self.get('quantity', '0')), decimal(self.get('asset_cost', '0'))
                fee = result.get('fee')
                fee_cost = decimal(fee) if fee is not None else qty * price * decimal(self.cfg['fee_reserve_rate'])
                cash_low = decimal(self.get('cash_lower', self.cfg['starting_capital']))
                cash_high = decimal(self.get('cash_upper', self.cfg['starting_capital']))
                gross = qty * price
                minimum_fee = decimal(fee) if fee is not None else ZERO
                if intent['side'] == 'buy':
                    cash_low, cash_high = cash_low - gross - fee_cost, cash_high - gross - minimum_fee
                else:
                    cash_low, cash_high = cash_low + gross - fee_cost, cash_high + gross - minimum_fee
                self.set('cash_lower', cash_low)
                self.set('cash_upper', cash_high)
                if intent['side'] == 'buy':
                    held, cost = held + qty, cost + qty * price + fee_cost
                else:
                    if qty > held + D('0.000000001'):
                        self.halt('sell_exceeds_accounted_holdings')
                    remaining = max(ZERO, held - qty)
                    cost = cost * remaining / held if held else ZERO
                    held = remaining
                self.set('quantity', held)
                self.set('asset_cost', cost)
                target, deviation = target_deviation(decimal(intent['reference']), price,
                    self.cfg['execution_target_multiplier'])
                if deviation >= D('0.0001'):
                    self._lesson('order.fill.price_vs_target', {'client_id': cid, 'broker_id': bid,
                        'target': target, 'fill': price, 'deviation_pct': deviation,
                        'threshold_pct': D('0.0001'), 'fee_basis': result['fee_basis']})
            self._append('EXECUTION_RECEIPT', {'client_id': cid, **result})
            self.set('execution_generation', int(self.get('execution_generation', 0)) + 1)

    def observe(self, kind, snapshot, **extra):
        with self.mutex, self.db:
            equity = decimal(snapshot['equity'])
            peak = max(equity, decimal(self.get('peak', self.cfg['starting_capital'])))
            floor, ratchet = capital_floor(self.cfg['starting_capital'],
                self.get('asset_cost', '0'), peak, self.get('ratchet', '250'))
            self.set('peak', peak)
            self.set('ratchet', ratchet)
            reserve = decimal(snapshot['gross_asset_value']) * (
                decimal(self.cfg['fee_reserve_rate']) + decimal(self.cfg['slippage_reserve_rate']))
            net = equity - reserve
            count = self.lesson_count()
            body = {'snapshot': snapshot, 'peak': peak, 'milestone_floor': ratchet,
                'floor': floor, 'net_liquidation_estimate': net,
                'asset_cost_including_fee_or_reserve': self.get('asset_cost', '0'),
                'lesson_count': count, 'risk_multiplier': risk_multiplier(count, self.cfg['lesson_risk_decay']),
                'halted': self.halted.is_set(), **extra}
            self._append(kind, body)
            return net, floor

    def close(self):
        with self.mutex:
            self.db.close()
            fcntl.flock(self.lock_file, fcntl.LOCK_UN)
            self.lock_file.close()


class Broker:
    def __init__(self, cfg, key, secret, journal):
        self.cfg, self.key, self.secret, self.journal = cfg, key, secret, journal
        self.symbol = cfg['symbol']
        self.session = None
        self.stream_ready = asyncio.Event()

    async def __aenter__(self):
        self.session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=8),
            trust_env=False, raise_for_status=False)
        return self

    async def __aexit__(self, *args):
        await self.session.close()

    async def http(self, method, url, **kwargs):
        try:
            async with self.session.request(method, url, allow_redirects=False, **kwargs) as response:
                if response.status >= 300:
                    raise Fault('broker_http_' + str(response.status))
                raw = await response.content.read(4 * 1024 * 1024 + 1)
                if len(raw) > 4 * 1024 * 1024:
                    raise Fault('broker_response_too_large')
                return decode(raw) if raw else {}
        except (aiohttp.ClientError, asyncio.TimeoutError):
            raise Fault('broker_transport_failure') from None

    async def frames(self, socket):
        async for message in socket:
            if message.type in (aiohttp.WSMsgType.TEXT, aiohttp.WSMsgType.BINARY):
                yield decode(message.data)
            elif message.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR):
                break
        raise Fault('broker_socket_closed')

    async def terminal(self, broker_id, timeout=15):
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            result = await self.order_info(broker_id)
            if result['status'] in TERMINAL:
                return result
            await asyncio.sleep(DURATION_POLL)
        raise Fault('order_reconciliation_timeout')


DURATION_POLL = 0.5  # Scheduling intervals only; no money passes through float.


class Alpaca(Broker):
    REST = 'https://api.alpaca.markets'
    DATA = 'wss://stream.data.alpaca.markets/v2/'
    ORDERS = 'wss://api.alpaca.markets/stream'

    def headers(self):
        return {'APCA-API-KEY-ID': self.key, 'APCA-API-SECRET-KEY': self.secret}

    async def rest(self, method, path, **kwargs):
        return await self.http(method, self.REST + path, headers=self.headers(), **kwargs)

    async def prepare(self):
        account = await self.rest('GET', '/v2/account')
        if account.get('currency') != 'USD' or account.get('status') != 'ACTIVE':
            raise Fault('active_USD_account_required')
        if account.get('trading_blocked') or decimal(account.get('short_market_value', '0')) != 0:
            raise Fault('blocked_or_short_account')
        if decimal(account.get('multiplier', '1')) != 1:
            raise Fault('disable_margin_use_cash_account')
        old = self.journal.get('broker_account_id')
        if old and old != account['id']:
            raise Fault('broker_account_identity_changed')
        with self.journal.mutex, self.journal.db:
            self.journal.set('broker_account_id', account['id'])
        asset = await self.rest('GET', '/v2/assets/' + urllib.parse.quote(self.symbol, safe=''))
        if asset.get('tradable') is not True or asset.get('fractionable') is not True:
            raise Fault('tradable_fractionable_equity_required')

    async def snapshot(self):
        account, positions = await asyncio.gather(self.rest('GET', '/v2/account'),
                                                 self.rest('GET', '/v2/positions'))
        if account['id'] != self.journal.get('broker_account_id'):
            raise Fault('account_identity_mismatch')
        if account.get('currency') != 'USD' or decimal(account.get('multiplier', '1')) != 1:
            raise Fault('account_scope_changed')
        foreign = [p['symbol'] for p in positions if p['symbol'] != self.symbol or p['side'] != 'long']
        if foreign:
            raise Fault('unsupported_or_external_position')
        qty = sum((decimal(p['qty']) for p in positions), ZERO)
        gross = sum((decimal(p['market_value']) for p in positions), ZERO)
        return {'at': now(), 'cash': decimal(account['cash']), 'quantity': qty,
            'equity': decimal(account['equity']), 'gross_asset_value': gross,
            'source': 'alpaca_REST_account_and_positions'}

    async def orders(self):
        return await self.rest('GET', '/v2/orders?status=open&limit=500')

    async def buy_allowed(self):
        clock = await self.rest('GET', '/v2/clock')
        if clock.get('is_open') is not True:
            raise Fault('equity_market_closed')

    async def submit(self, cid, side, qty):
        row = await self.rest('POST', '/v2/orders', json={'symbol': self.symbol,
            'qty': format(qty, 'f'), 'side': side, 'type': 'market',
            'time_in_force': 'day', 'client_order_id': cid})
        if row.get('client_order_id') != cid:
            raise Fault('order_identity_mismatch')
        return row['id']

    async def find_order(self, cid):
        row = await self.rest('GET', '/v2/orders:by_client_order_id', params={'client_order_id': cid})
        if row.get('client_order_id') != cid:
            raise Fault('recovery_order_identity_mismatch')
        return row['id']

    async def order_info(self, bid):
        row = await self.rest('GET', '/v2/orders/' + urllib.parse.quote(bid, safe=''))
        qty = decimal(row.get('filled_qty', '0'))
        return {'broker_id': row['id'], 'status': row['status'], 'quantity': qty,
            'price': decimal(row.get('filled_avg_price') or '0'), 'fee': None,
            'fee_basis': 'fee_unknown_conservative_configured_reserve', 'at': now()}

    async def cancel_all(self):
        await self.rest('DELETE', '/v2/orders')

    async def bars(self):
        async with self.session.ws_connect(self.DATA + self.cfg['alpaca_feed'], heartbeat=20,
                max_msg_size=2 * 1024 * 1024) as ws:
            await ws.send_json({'action': 'auth', 'key': self.key, 'secret': self.secret})
            subscribed = False
            async for packet in self.frames(ws):
                for row in packet:
                    if row.get('T') == 'error':
                        raise Fault('alpaca_market_stream_rejected')
                    if row.get('T') == 'success' and row.get('msg') == 'authenticated':
                        await ws.send_json({'action': 'subscribe', 'bars': [self.symbol]})
                    if row.get('T') == 'subscription':
                        if self.symbol not in row.get('bars', []):
                            raise Fault('alpaca_bar_subscription_missing')
                        subscribed = True
                    if row.get('T') == 'b' and row.get('S') == self.symbol and subscribed:
                        yield {'at': row['t'], 'price': decimal(row['c']), 'source': 'alpaca_bar'}

    async def account_stream(self):
        async with self.session.ws_connect(self.ORDERS, heartbeat=20) as ws:
            await ws.send_json({'action': 'auth', 'key': self.key, 'secret': self.secret})
            async for row in self.frames(ws):
                if row.get('stream') == 'authorization':
                    if row.get('data', {}).get('status') != 'authorized':
                        raise Fault('alpaca_order_stream_auth_failed')
                    await ws.send_json({'action': 'listen', 'data': {'streams': ['trade_updates']}})
                elif row.get('stream') == 'listening':
                    if 'trade_updates' not in row.get('data', {}).get('streams', []):
                        raise Fault('alpaca_order_subscription_missing')
                    self.stream_ready.set()
                elif row.get('stream') == 'trade_updates':
                    data = row['data']
                    # Persist selected nonsecret fields, never arbitrary account payloads.
                    self.journal.append('ORDER_STREAM', {'event': data['event'],
                        'broker_id': data['order']['id'], 'status': data['order']['status'],
                        'filled_qty': decimal(data['order'].get('filled_qty', '0'))})


class Kraken(Broker):
    REST = 'https://api.kraken.com'
    PUBLIC = 'wss://ws.kraken.com/v2'
    PRIVATE = 'wss://ws-auth.kraken.com/v2'

    def __init__(self, *args):
        super().__init__(*args)
        self.balance = {}
        self.balance_ready = asyncio.Event()
        self.last_private_message = 0
        self.sequence = None
        self.ws = None
        self.ws_pending = {}
        self.req = 0
        self.pair = None
        self.base_code = None
        self.base_name = self.symbol.split('/')[0]
        self.token = None

    def signed(self, path, data):
        # Nonce is durable and serialized across the two broker threads/process restarts.
        with self.journal.mutex, self.journal.db:
            nonce = max(time.time_ns() // 1000, int(self.journal.get('nonce', 0)) + 1)
            self.journal.set('nonce', nonce)
        body = urllib.parse.urlencode({'nonce': str(nonce), **data})
        digest = hashlib.sha256((str(nonce) + body).encode()).digest()
        try:
            secret = base64.b64decode(self.secret, validate=True)
        except ValueError:
            raise Fault('kraken_secret_must_be_base64') from None
        signature = base64.b64encode(hmac.new(secret, path.encode() + digest, hashlib.sha512).digest()).decode()
        return body, {'API-Key': self.key, 'API-Sign': signature,
                      'Content-Type': 'application/x-www-form-urlencoded'}

    async def rest(self, endpoint, data=None, private=True):
        path = '/0/' + ('private/' if private else 'public/') + endpoint
        if private:
            # Separate per-process nonce serialization includes request completion so a
            # later nonce cannot arrive before an earlier one on another thread.
            async with thread_gate(self.journal.kraken_gate):
                body, headers = self.signed(path, data or {})
                row = await self.http('POST', self.REST + path, data=body, headers=headers)
        else:
            row = await self.http('GET', self.REST + path, params=data or {})
        if row.get('error'):
            raise Fault('kraken_request_rejected')
        return row['result']

    async def prepare(self):
        pairs = await self.rest('AssetPairs', private=False)
        wanted = ('XBT' if self.base_name == 'BTC' else self.base_name) + '/USD'
        matches = [(code, p) for code, p in pairs.items() if p.get('wsname') == wanted]
        if len(matches) != 1:
            raise Fault('kraken_pair_not_supported')
        self.pair, info = matches[0]
        if info.get('status') != 'online':
            raise Fault('kraken_pair_not_online')
        self.base_code = info['base']
        self.qty_step = D('1').scaleb(-int(info['lot_decimals']))
        self.minimum = decimal(info['ordermin'])
        self.cost_minimum = decimal(info.get('costmin', '0'))
        token = await self.rest('GetWebSocketsToken')
        self.token = token['token']
        if await self.rest('OpenPositions'):
            raise Fault('kraken_margin_positions_unsupported')

    def validate_balances(self, balances):
        allowed = {'ZUSD', 'USD', self.base_code, self.base_name,
                   'XBT' if self.base_name == 'BTC' else self.base_name}
        for asset, amount in balances.items():
            if decimal(amount) < 0 or (decimal(amount) != 0 and asset not in allowed):
                raise Fault('kraken_requires_dedicated_USD_and_configured_asset_spot_account')

    async def snapshot(self):
        # Sample authenticated balance state and a fresh public bid. Private streams
        # avoid 1 Hz private REST polling that can exhaust Kraken's rate budget.
        if self.balance_ready.is_set():
            if time.monotonic() - self.last_private_message > 10:
                raise Fault('kraken_private_stream_stale')
            balances = dict(self.balance)
        else:
            balances = await self.rest('Balance')
        self.validate_balances(balances)
        cash = sum((decimal(v) for a, v in balances.items() if a in ('USD', 'ZUSD')), ZERO)
        qty = sum((decimal(v) for a, v in balances.items() if a not in ('USD', 'ZUSD')), ZERO)
        ticker = await self.rest('Ticker', {'pair': self.pair}, private=False)
        bid = decimal(next(iter(ticker.values()))['b'][0])
        if bid <= 0:
            raise Fault('invalid_kraken_bid')
        gross = qty * bid
        return {'at': now(), 'cash': cash, 'quantity': qty, 'equity': cash + gross,
                'gross_asset_value': gross, 'source': 'kraken_balances_and_public_bid'}

    async def orders(self):
        return list((await self.rest('OpenOrders'))['open'].values())

    async def buy_allowed(self):
        status = await self.rest('SystemStatus', private=False)
        if status.get('status') != 'online' or self.ws is None or self.ws.closed:
            raise Fault('kraken_exchange_or_authenticated_socket_unavailable')

    async def request_ws(self, method, params):
        if self.ws is None or self.ws.closed:
            raise Fault('kraken_authenticated_socket_unavailable')
        self.req += 1
        rid = self.req
        future = asyncio.get_running_loop().create_future()
        self.ws_pending[rid] = future
        try:
            await self.ws.send_str(wire({'method': method, 'params': {**params, 'token': self.token}, 'req_id': rid}))
            row = await asyncio.wait_for(future, timeout=8)
            if row.get('success') is not True:
                raise Fault('kraken_websocket_request_rejected')
            return row['result']
        finally:
            self.ws_pending.pop(rid, None)

    async def submit(self, cid, side, qty):
        qty = (qty / self.qty_step).to_integral_value(rounding=ROUND_DOWN) * self.qty_step
        if qty < self.minimum:
            raise Fault('kraken_order_below_minimum')
        result = await self.request_ws('add_order', {'order_type': 'market', 'side': side,
            'order_qty': qty, 'symbol': self.symbol, 'cl_ord_id': cid, 'fee_preference': 'quote',
            'deadline': (datetime.now(timezone.utc) + timedelta(seconds=5)).isoformat(timespec='milliseconds')})
        return result['order_id']

    async def find_order(self, cid):
        opened = await self.rest('OpenOrders', {'cl_ord_id': cid})
        closed = await self.rest('ClosedOrders', {'cl_ord_id': cid})
        matches = {**opened['open'], **closed['closed']}
        if len(matches) != 1:
            raise Fault('kraken_recovery_order_missing_or_ambiguous')
        return next(iter(matches))

    async def order_info(self, bid):
        rows = await self.rest('QueryOrders', {'txid': bid})
        if bid not in rows:
            raise Fault('kraken_order_reconciliation_missing')
        row = rows[bid]
        return {'broker_id': bid, 'status': row['status'], 'quantity': decimal(row['vol_exec']),
                'price': decimal(row['price']), 'fee': decimal(row['fee']),
                'fee_basis': 'authoritative_broker_order_aggregate', 'at': now()}

    async def cancel_all(self):
        await self.rest('CancelAll')

    async def bars(self):
        async with self.session.ws_connect(self.PUBLIC, heartbeat=20, max_msg_size=4 * 1024 * 1024) as ws:
            await ws.send_json({'method': 'subscribe', 'params': {'channel': 'ohlc',
                'symbol': [self.symbol], 'interval': 1, 'snapshot': False}})
            async for row in self.frames(ws):
                if row.get('success') is False:
                    raise Fault('kraken_market_subscription_rejected')
                if row.get('channel') == 'ohlc' and row.get('type') == 'update':
                    for bar in row['data']:
                        if bar['symbol'] == self.symbol:
                            yield {'at': bar['interval_begin'], 'price': decimal(bar['close']),
                                   'source': 'kraken_ohlc_update'}

    async def account_stream(self):
        async with self.session.ws_connect(self.PRIVATE, heartbeat=5, max_msg_size=4 * 1024 * 1024) as ws:
            self.ws = ws
            for channel in ('balances', 'executions'):
                await ws.send_json({'method': 'subscribe', 'params': {'channel': channel,
                    'token': self.token, **({'snapshot': True} if channel == 'balances' else {})}})
            try:
                async for row in self.frames(ws):
                    self.last_private_message = time.monotonic()
                    future = self.ws_pending.get(row.get('req_id'))
                    if future is not None and not future.done():
                        future.set_result(row)
                    if row.get('success') is False:
                        raise Fault('kraken_private_stream_rejected')
                    if row.get('channel') == 'balances':
                        sequence = row.get('sequence')
                        if row.get('type') == 'snapshot':
                            self.balance.clear()
                        elif self.sequence is not None and sequence != self.sequence + 1:
                            raise Fault('kraken_balance_sequence_gap')
                        self.sequence = sequence
                        for asset in row['data']:
                            for wallet in asset.get('wallets', []):
                                if decimal(wallet['balance']) != 0 and (wallet['type'] != 'spot' or wallet['id'] != 'main'):
                                    raise Fault('kraken_nonspot_wallet_balance')
                            if asset.get('wallet_type', 'spot') != 'spot':
                                raise Fault('kraken_nonspot_wallet_activity')
                            self.balance[asset['asset']] = decimal(asset['balance'])
                            if row.get('type') == 'update' and asset.get('type') not in ('trade',):
                                self.journal.halt('external_account_movement_' + asset.get('type', 'unknown'))
                        self.validate_balances(self.balance)
                        self.balance_ready.set()
                        self.stream_ready.set()
                    elif row.get('channel') == 'executions':
                        for event in row['data']:
                            self.journal.append('ORDER_STREAM', {key: event[key] for key in
                                ('order_id', 'cl_ord_id', 'exec_type', 'order_status', 'cum_qty', 'avg_price') if key in event})
            finally:
                self.ws = None
                self.balance_ready.clear()
                for future in list(self.ws_pending.values()):
                    if not future.done():
                        future.set_exception(Fault('kraken_socket_lost_during_request'))


class thread_gate:
    """Cancellation-safe acquisition of a cross-thread lock without blocking the loop."""
    def __init__(self, lock):
        self.lock = lock

    async def __aenter__(self):
        while not self.lock.acquire(blocking=False):
            await asyncio.sleep(0.01)
        return self

    async def __aexit__(self, *args):
        self.lock.release()


def broker(cfg, key, secret, journal):
    return (Alpaca if cfg['broker'] == 'alpaca' else Kraken)(cfg, key, secret, journal)


def account_mismatch(snapshot, journal):
    if not journal.get('initialized'):
        return False
    quantity = decimal(snapshot['quantity'])
    expected = decimal(journal.get('quantity', '0'))
    cash = decimal(snapshot['cash'])
    low = decimal(journal.get('cash_lower', journal.cfg['starting_capital']))
    high = decimal(journal.get('cash_upper', journal.cfg['starting_capital']))
    return abs(quantity - expected) > D('0.000000001') or cash < low - D('0.01') or cash > high + D('0.01')


def indicators(root, at):
    """Re-read monthly/event sources on every strategy tick; bounded file size.

    Empty/new project => HOLD, while networking and risk monitoring still run.
    Post-2000 supplied publication dates control visibility. Pre-2000 rows use
    observation dates under the owner's retained historical alignment policy.
    """
    months, events, future_in_month = {}, [], False
    for filename in ('macro.jsonl', 'geopolitical.jsonl'):
        path = root / filename
        if not path.is_file() or path.stat().st_size > 16 * 1024 * 1024:
            raise Fault('missing_or_oversized_' + filename)
        for line in path.read_text().splitlines():
            if not line.strip():
                continue
            row = decode(line)
            if filename == 'macro.jsonl':
                if row['series'] != 'INDPRO':
                    continue
                obs = timestamp(row['observation_date'] + ('T00:00:00+00:00' if len(row['observation_date']) == 10 else ''))
                supplied = row.get('available_at', row.get('publication_date'))
                available = obs if obs.year < 2000 or not supplied else timestamp(supplied)
                if available < obs:
                    raise Fault('macro_availability_before_observation')
                if obs <= at and available <= at:
                    month = obs.strftime('%Y-%m')
                    if month not in months or available > months[month][0]:
                        months[month] = (available, decimal(row['value']))
            else:
                event_at = timestamp(row['at'])
                if event_at.strftime('%Y-%m') == at.strftime('%Y-%m'):
                    if event_at <= at:
                        events.append(decimal(row['tone']))
                    else:
                        future_in_month = True
    keys = sorted(months)[-4:]
    indexes = [int(k[:4]) * 12 + int(k[5:]) for k in keys]
    values = [months[k][1] for k in keys] if len(keys) == 4 and indexes[-1] - indexes[0] == 3 else []
    tone = min(events) if events else (None if future_in_month else ZERO)
    return {'indpro': values, 'tone': tone, 'month': keys[-1] if keys else '',
            'macro_visibility_policy': 'supplied_available_at_else_observation_date'}


async def reconcile_pending(adapter, journal):
    with journal.mutex:
        pending = journal.db.execute("SELECT client_id,broker_id FROM intents WHERE status='pending'").fetchall()
    for cid, bid in pending:
        if not bid:
            try:
                bid = await adapter.find_order(cid)
                with journal.mutex, journal.db:
                    journal.db.execute('UPDATE intents SET broker_id=? WHERE client_id=?', (bid, cid))
                    journal._append('RECOVERED_ORDER_ID', {'client_id': cid, 'broker_id': bid})
            except Fault:
                journal.append('AMBIGUOUS_INTENT', {'client_id': cid, 'policy': 'never_resubmit'})
                continue
        if bid:
            try:
                row = await adapter.terminal(bid, timeout=10)
                journal.receipt(cid, row)
            except Fault:
                journal.append('RECONCILIATION_UNCONFIRMED', {'client_id': cid, 'broker_id': bid})
        else:
            journal.append('AMBIGUOUS_INTENT', {'client_id': cid, 'policy': 'never_resubmit'})


async def execute(adapter, journal, side, quantity, reference, month='', emergency=False):
    if quantity <= 0:
        raise Fault('nonpositive_order_quantity')
    cid = journal.intent(side, quantity, reference, month, emergency)
    # Persist the intent first; submission is NEVER retried after a timeout.
    bid = await adapter.submit(cid, side, quantity)
    with journal.mutex, journal.db:
        journal.db.execute('UPDATE intents SET broker_id=? WHERE client_id=?', (bid, cid))
        journal._append('ORDER_ACK', {'client_id': cid, 'broker_id': bid})
    row = await adapter.terminal(bid)
    journal.receipt(cid, row)
    expected = quantity
    if isinstance(adapter, Kraken):
        expected = (quantity / adapter.qty_step).to_integral_value(rounding=ROUND_DOWN) * adapter.qty_step
    if (row['status'] not in ('filled', 'closed') or decimal(row['quantity']) <= 0
            or abs(decimal(row['quantity']) - expected) > D('0.000000001')):
        raise Fault('entry_or_exit_not_fully_filled')
    return row


async def liquidate(adapter, journal, reason):
    """Cancel, reconcile, then close the managed position and verify flatness.

    Account scope errors and unsellable dust are UNCONFIRMED, never false success.
    Retrying this function reconciles the book first; pending submissions are not
    replayed. A failed/ambiguous liquidation is not automatically repeated.
    """
    journal.halt(reason)
    # Cancel immediately, independently of the strategy's fill-wait lock. This
    # unblocks terminal reconciliation before serializing emergency submissions.
    try:
        await adapter.cancel_all()
    except Exception as exc:
        journal.append('LIQUIDATION_UNCONFIRMED', {'reason': reason, 'stage': 'cancel_open_orders',
            'error_type': type(exc).__name__})
        log('LIQUIDATION_UNCONFIRMED', stage='cancel_open_orders')
        return False
    async with thread_gate(journal.order_gate):
        try:
            await reconcile_pending(adapter, journal)
            if await adapter.orders():
                raise Fault('cancel_not_reconciled')
            with journal.mutex:
                unresolved = journal.db.execute("SELECT count(*) FROM intents WHERE status='pending'").fetchone()[0]
            with journal.mutex:
                pending_sells = journal.db.execute(
                    "SELECT body FROM intents WHERE status='pending'").fetchall()
            if any(decode(r[0])['side'] == 'sell' for r in pending_sells):
                raise Fault('prior_sell_ambiguous_no_duplicate_liquidation')
            # Unknown submission cannot be proven absent by open-orders alone: it
            # may have filled. Close observed inventory, retain unresolved audit state.
            snap = await adapter.snapshot()
            qty = decimal(snap['quantity'])
            if qty > 0:
                reference = decimal(snap['gross_asset_value']) / qty
                await execute(adapter, journal, 'sell', qty, reference, emergency=True)
            snap = await adapter.snapshot()
            if decimal(snap['quantity']) != 0 or await adapter.orders() or unresolved:
                raise Fault('flatness_or_prior_submission_unconfirmed')
            journal.observe('LIQUIDATION_CONFIRMED', snap, reason=reason)
            log('LIQUIDATION_CONFIRMED')
            return True
        except Exception as exc:
            journal.append('LIQUIDATION_UNCONFIRMED', {'reason': reason, 'error_type': type(exc).__name__,
                'fault': str(exc) if isinstance(exc, Fault) else 'internal_error'})
            log('LIQUIDATION_UNCONFIRMED', error_type=type(exc).__name__)
            return False


class RiskWorker(threading.Thread):
    """A separate event loop/session continues when the strategy loop stalls."""
    def __init__(self, cfg, key, secret, journal):
        super().__init__(name='alphabot-independent-risk', daemon=False)
        self.cfg, self.key, self.secret, self.journal = cfg, key, secret, journal
        self.stop_event = threading.Event()
        self.ready = threading.Event()
        self.last_sample = 0
        self.failed = False

    def run(self):
        getcontext().prec = 50
        try:
            asyncio.run(self.monitor())
        except BaseException:
            self.failed = True
            self.journal.halted.set()
            with contextlib.suppress(Exception):
                self.journal.halt('risk_worker_failed')
            self.ready.set()

    async def monitor(self):
        async with broker(self.cfg, self.key, self.secret, self.journal) as adapter:
            await adapter.prepare()
            # Kraken emergency submissions require its own authenticated socket.
            stream = asyncio.create_task(adapter.account_stream()) if isinstance(adapter, Kraken) else None
            if stream:
                await asyncio.wait_for(adapter.balance_ready.wait(), 10)
            due = time.monotonic()
            try:
                while not self.stop_event.is_set():
                    if stream and stream.done():
                        stream.result()
                        raise Fault('risk_private_stream_ended')
                    try:
                        generation = self.journal.get('execution_generation', 0)
                        snap = await adapter.snapshot()
                        self.last_sample = time.monotonic()
                        net, floor = self.journal.observe('RISK_SAMPLE', snap)
                        context = indicators(self.journal.root, datetime.now(timezone.utc))
                        self.ready.set()
                        with self.journal.mutex:
                            stable = generation == self.journal.get('execution_generation', 0) and not self.journal.order_gate.locked()
                            mismatch = stable and account_mismatch(snap, self.journal)
                        reason = ('account_reconciliation_mismatch' if mismatch else
                            'equity_below_floor' if net < floor else
                            'geopolitical_risk_override' if context['tone'] is not None and context['tone'] < -5 else
                            'durable_halt' if self.journal.halted.is_set() else None)
                        if reason:
                            await liquidate(adapter, self.journal, reason)
                            return
                    except Exception as exc:
                        self.ready.set()
                        await liquidate(adapter, self.journal, 'risk_sample_failed_' + type(exc).__name__)
                        self.failed = True
                        return
                    due += 1
                    delay = due - time.monotonic()
                    if delay < -1:
                        await liquidate(adapter, self.journal, 'risk_sampling_deadline_missed')
                        self.failed = True
                        return
                    await asyncio.sleep(max(0, delay))
            finally:
                if stream:
                    stream.cancel()
                    await asyncio.gather(stream, return_exceptions=True)


async def trading(root, cfg, key, secret, journal, check=False, duration=None):
    async with broker(cfg, key, secret, journal) as adapter:
        await adapter.prepare()
        initial = await adapter.snapshot()
        journal.observe('PREFLIGHT', initial)
        if check:
            log('READ_ONLY_CHECK', snapshot=initial, halted=journal.halted.is_set(),
                unresolved_intents=journal.db.execute("SELECT count(*) FROM intents WHERE status='pending'").fetchone()[0])
            return
        # A new project cannot import unknown historical acquisition cost silently.
        if not journal.get('initialized'):
            if decimal(initial['quantity']) != 0 or await adapter.orders():
                raise Fault('first_start_requires_flat_dedicated_account')
            if abs(decimal(initial['cash']) - decimal(cfg['starting_capital'])) > D('0.01'):
                raise Fault('starting_capital_does_not_match_account_cash')
            with journal.mutex, journal.db:
                journal.set('initialized', True)
                journal.set('quantity', '0')
                journal.set('asset_cost', '0')
                journal.set('cash_lower', cfg['starting_capital'])
                journal.set('cash_upper', cfg['starting_capital'])
        elif abs(decimal(initial['quantity']) - decimal(journal.get('quantity', '0'))) > D('0.000000001'):
            journal.halt('restart_holdings_reconciliation_mismatch')
        private = asyncio.create_task(adapter.account_stream(), name='authenticated-account-stream')
        risk = RiskWorker(cfg, key, secret, journal)
        risk.start()
        stopping = asyncio.Event()
        loop = asyncio.get_running_loop()
        for sig in (signal.SIGINT, signal.SIGTERM):
            with contextlib.suppress(NotImplementedError):
                loop.add_signal_handler(sig, stopping.set)
        started = time.monotonic()
        async def consume():
            await asyncio.wait_for(adapter.stream_ready.wait(), 10)
            deadline = time.monotonic() + 15
            while not risk.ready.is_set():
                if time.monotonic() >= deadline:
                    raise Fault('risk_worker_startup_timeout')
                await asyncio.sleep(0.01)
            async for bar in adapter.bars():
                price = decimal(bar['price'])
                age = D(str(time.time())) - D(str(timestamp(bar['at']).timestamp()))
                if price <= 0 or age < -5 or age > cfg['market_data_max_age_seconds']:
                    raise Fault('stale_future_or_invalid_bar')
                snap = await adapter.snapshot()
                net, floor = journal.observe('MARKET_TICK', snap, bar=bar)
                if journal.halted.is_set() or stopping.is_set():
                    return
                if net < floor:
                    await liquidate(adapter, journal, 'strategy_preflight_floor_breach')
                    return
                context = {**indicators(root, datetime.now(timezone.utc)), 'cash': snap['cash'],
                    'last_entry_month': journal.get('last_entry_month'),
                    'lesson_count': journal.lesson_count(), 'lesson_risk_decay': cfg['lesson_risk_decay']}
                decision = decide(context)
                journal.append('DECISION', {'context': context, 'decision': decision})
                if decision['action'] == 'liquidate':
                    await liquidate(adapter, journal, decision['reason'])
                    return
                if decision['action'] != 'buy':
                    continue
                async with thread_gate(journal.order_gate):
                    if journal.halted.is_set() or risk.failed or time.monotonic() - risk.last_sample > 3:
                        raise Fault('risk_health_blocks_order')
                    await adapter.buy_allowed()
                    if await adapter.orders():
                        raise Fault('unmanaged_open_orders')
                    fresh = await adapter.snapshot()
                    if account_mismatch(fresh, journal):
                        raise Fault('account_movement_or_holdings_mismatch')
                    net, floor = journal.observe('ORDER_PREFLIGHT', fresh)
                    budget = min(decimal(decision['budget']), decimal(cfg['max_order_usd']), decimal(fresh['cash']))
                    drag = decimal(cfg['fee_reserve_rate']) + decimal(cfg['slippage_reserve_rate'])
                    # Reserve modeled entry+exit costs; market fills can still gap.
                    projected_floor, _ = capital_floor(cfg['starting_capital'],
                        decimal(journal.get('asset_cost', '0')) + budget,
                        journal.get('peak', cfg['starting_capital']), journal.get('ratchet', '250'))
                    if net - budget * drag * 2 < projected_floor:
                        journal.append('ORDER_BLOCKED', {'reason': 'projected_floor', 'budget': budget})
                        continue
                    qty = (budget / (price * (1 + drag))).quantize(D('0.00000001'), rounding=ROUND_DOWN)
                    if qty <= 0 or (isinstance(adapter, Kraken) and (qty * price < adapter.cost_minimum or qty < adapter.minimum)):
                        journal.append('ORDER_BLOCKED', {'reason': 'venue_minimum_notional'})
                        continue
                    await execute(adapter, journal, 'buy', qty, price, context['month'])
        consumer = asyncio.create_task(consume(), name='live-market-execution')
        try:
            while not stopping.is_set():
                if private.done():
                    private.result()
                    raise Fault('account_stream_ended')
                if consumer.done():
                    consumer.result()
                    break
                if not risk.is_alive() or risk.failed:
                    break
                if duration is not None and time.monotonic() - started >= duration:
                    break
                await asyncio.sleep(0.1)
        except Exception as exc:
            journal.halt('execution_failure_' + type(exc).__name__)
            log('EXECUTION_FAILURE', error_type=type(exc).__name__,
                fault=str(exc) if isinstance(exc, Fault) else 'internal_error')
        finally:
            # Stop admission before canceling a possibly in-flight strategy request.
            journal.halt('session_shutdown')
            consumer.cancel()
            await asyncio.gather(consumer, return_exceptions=True)
            risk.stop_event.set()
            # Risk gets first ownership of emergency reconciliation if already active.
            while risk.is_alive():
                await asyncio.sleep(0.05)
            await liquidate(adapter, journal, 'session_shutdown')
            private.cancel()
            await asyncio.gather(private, return_exceptions=True)
            journal.verify()


async def rearm(cfg, key, secret, journal, reason):
    async with broker(cfg, key, secret, journal) as adapter:
        await adapter.prepare()
        await reconcile_pending(adapter, journal)
        snapshot = await adapter.snapshot()
        net, floor = journal.observe('REARM_PREFLIGHT', snapshot)
        with journal.mutex:
            unresolved = journal.db.execute("SELECT count(*) FROM intents WHERE status='pending'").fetchone()[0]
        if (unresolved or await adapter.orders() or decimal(snapshot['quantity']) != 0
                or account_mismatch(snapshot, journal) or net < floor):
            raise Fault('rearm_requires_reconciled_flat_account_above_floor')
        with journal.mutex, journal.db:
            old_halt = journal.get('halt')
            journal._append('OPERATOR_REARM', {'reason': reason, 'prior_halt': old_halt,
                'preserved_peak': journal.get('peak'), 'preserved_ratchet': journal.get('ratchet'),
                'preserved_lesson_count': journal.lesson_count()})
            journal.set('halt', None)
            journal.halted.clear()
        log('REARMED', preserved_lesson_count=journal.lesson_count())


def initialize(root, name, symbol):
    root.mkdir(parents=True, exist_ok=True, mode=0o700)
    if (root / 'config.json').exists():
        raise Fault('refusing_to_overwrite_existing_project')
    cfg = {'broker': name, 'symbol': symbol or ('SPY' if name == 'alpaca' else 'BTC/USD'),
           'starting_capital': '500', 'max_order_usd': '100', 'fee_reserve_rate': '0.005',
           'slippage_reserve_rate': '0.01', 'lesson_risk_decay': '0.95',
           'execution_target_multiplier': '1', 'dedicated_account': True}
    (root / 'config.json').write_text(json.dumps(cfg, indent=2) + '\n')
    (root / 'macro.jsonl').touch(exist_ok=False)
    (root / 'geopolitical.jsonl').touch(exist_ok=False)
    log('PROJECT_INITIALIZED', project=str(root), policy='empty_INDPRO_means_HOLD')


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('command', choices=('init', 'check', 'run', 'verify', 'rearm'))
    parser.add_argument('--project', required=True, type=Path)
    parser.add_argument('--broker', choices=('alpaca', 'kraken'), default='alpaca')
    parser.add_argument('--symbol')
    parser.add_argument('--live', action='store_true', help='explicitly enable real-money order submission')
    parser.add_argument('--acknowledge-halt', action='store_true')
    parser.add_argument('--reason', help='audit reason for an explicit rearm')
    parser.add_argument('--duration', type=int, help='end session after N seconds; flatten on exit')
    args = parser.parse_args()
    root = args.project.expanduser().resolve()
    if args.command == 'init':
        initialize(root, args.broker, args.symbol)
        return 0
    cfg = configuration(root)
    if args.command == 'run' and not args.live:
        raise Fault('run_requires_explicit_--live')
    if args.command == 'rearm' and (not args.acknowledge_halt or not args.reason or len(args.reason) > 200):
        raise Fault('rearm_requires_--acknowledge-halt_and_--reason_up_to_200_characters')
    if args.duration is not None and args.duration <= 0:
        raise Fault('positive_duration_required')
    if args.command == 'verify':
        db_path = root / 'native-live.sqlite3'
        db = sqlite3.connect(db_path.as_uri() + '?mode=ro', uri=True)
        try:
            parent = '0' * 64
            db.execute('BEGIN')
            for at, kind, body, previous, digest in db.execute(
                    'SELECT at,kind,body,parent,hash FROM events ORDER BY id'):
                if previous != parent or hashlib.sha256(encode([at, kind, body, previous]).encode()).hexdigest() != digest:
                    raise Fault('journal_integrity_failure')
                parent = digest
            if db.execute('PRAGMA quick_check').fetchone()[0] != 'ok':
                raise Fault('sqlite_integrity_failure')
            log('JOURNAL_VERIFIED', event_count=db.execute('SELECT count(*) FROM events').fetchone()[0])
            return 0
        finally:
            db.close()
    key = os.environ.get('LIVE_BROKER_API_KEY')
    secret = os.environ.get('LIVE_BROKER_SECRET_KEY')
    if not key or not secret:
        raise Fault('LIVE_BROKER_API_KEY_and_LIVE_BROKER_SECRET_KEY_required')
    journal = Journal(root, cfg, hashlib.sha256(key.encode()).hexdigest())
    journal.kraken_gate = threading.Lock()
    try:
        if args.command == 'rearm':
            asyncio.run(rearm(cfg, key, secret, journal, args.reason))
        elif args.command == 'verify':
            journal.verify()
            log('JOURNAL_VERIFIED', event_count=journal.db.execute('SELECT count(*) FROM events').fetchone()[0])
        else:
            asyncio.run(trading(root, cfg, key, secret, journal, check=args.command == 'check', duration=args.duration))
        return 2 if journal.halted.is_set() else 0
    finally:
        journal.close()


if __name__ == '__main__':
    try:
        raise SystemExit(main())
    except (Fault, ValueError, KeyError, OSError) as exc:
        log('STARTUP_FAILED', error_type=type(exc).__name__,
            fault=str(exc) if isinstance(exc, Fault) else 'invalid_input_or_system_error')
        raise SystemExit(2)