"""Bounded read-only observations, with no manufactured futures adapter/entitlement."""
import asyncio
import json
from datetime import datetime,timezone
from verify_live_feeds_standalone import load_private_credentials,account_credentials,read_json,validate_quotes,Rejected
from infinbot_common import strict_json,instant

CRYPTO_STREAM='wss://sotality.com/v1beta3/crypto/us'


async def crypto_stream(settings,*,connect=None,timeout=15):
    if connect is None:
        return {'status':'PROXY_CRYPTO_WEBSOCKET_ROUTE_UNVERIFIED','broker_orders_submitted':0,
                'direct_alpaca_calls':0,'reason':'Proxy WebSocket route/handshake contract not supplied; no guessed live connection'}
    _,headers=account_credentials(settings,'paper')
    received={};subscription=False;authenticated=False
    try:
        async with connect(CRYPTO_STREAM,proxy=None,open_timeout=5,close_timeout=2,max_size=65536,
                           max_queue=4,compression=None) as ws:
            # Injected test transport only: no client broker authentication payload.
            deadline=asyncio.get_running_loop().time()+timeout
            while asyncio.get_running_loop().time()<deadline:
                data=strict_json(await asyncio.wait_for(ws.recv(),timeout=max(.01,deadline-asyncio.get_running_loop().time())))
                if not isinstance(data,list):raise Rejected('Malformed stream envelope')
                for event in data:
                    if not isinstance(event,dict) or event.get('T')=='error':raise Rejected('Stream rejected')
                    if event.get('T')=='success' and event.get('msg')=='authenticated':
                        authenticated=True;await ws.send(json.dumps({'action':'subscribe','quotes':['BTC/USD','ETH/USD']}))
                    elif event.get('T')=='subscription':
                        if not authenticated or not {'BTC/USD','ETH/USD'}<=set(event.get('quotes',[])):raise Rejected('Subscription not acknowledged')
                        subscription=True
                    elif event.get('T')=='q':
                        if not subscription:raise Rejected('Quote arrived before acknowledged subscription')
                        symbol=event.get('S')
                        if symbol not in ('BTC/USD','ETH/USD'):continue
                        observation=validate_quotes({'quotes':{symbol:{'bp':event.get('bp'),'ap':event.get('ap'),'t':event.get('t')}}},[symbol],datetime.now(timezone.utc))[symbol]
                        previous=received.get(symbol)
                        if previous and instant(observation['provider_timestamp'])<=instant(previous['provider_timestamp']):raise Rejected('Non-increasing stream timestamps')
                        received[symbol]=observation
                if set(received)=={'BTC/USD','ETH/USD'}:
                    return {'status':'FRESH_CRYPTO_STREAM_OBSERVED','authenticated':True,'subscription_acknowledged':True,
                            'symbols_observed':sorted(received),'provider_timestamps':{k:v['provider_timestamp'] for k,v in received.items()},
                            'persistent_worker_subscription_verified':False,'broker_orders_submitted':0}
    except Exception:
        return {'status':'CRYPTO_STREAM_UNVERIFIED','authenticated':authenticated,'subscription_acknowledged':subscription,'broker_orders_submitted':0}
    return {'status':'CRYPTO_STREAM_UNVERIFIED','broker_orders_submitted':0}


def broker_positions(settings):
    # Extend only the recovered GET endpoint allowlist, never add an order route.
    _,headers=account_credentials(settings,'paper')
    return read_json('https://sotality.com/v2/positions',headers)


def reconcile(local_positions,broker_positions):
    from infinbot_common import number
    if not isinstance(broker_positions,list):raise Rejected('Broker position array required')
    local={};broker={}
    for row in local_positions:
        symbol=row['symbol']
        if symbol in local:raise Rejected('Duplicate local position')
        local[symbol]=number(row['quantity'])
    for row in broker_positions:
        if not isinstance(row,dict) or not isinstance(row.get('symbol'),str):raise Rejected('Invalid broker position')
        symbol=row['symbol']
        if row.get('asset_class')=='crypto' and '/' not in symbol:
            allowed={'BTCUSD':'BTC/USD','ETHUSD':'ETH/USD'}
            if symbol not in allowed:raise Rejected('Unsupported broker symbol mapping')
            symbol=allowed[symbol]
        if symbol in broker:raise Rejected('Duplicate broker position')
        broker[symbol]=number(row['qty'])
    return {'status':'POSITION_RECONCILIATION_MATCH' if local==broker else 'POSITION_RECONCILIATION_MISMATCH',
            'local_position_count':len(local),'broker_position_count':len(broker),'scope':'Entire selected local and paper-broker portfolios',
            'liquidation_verified':False,'live_clearance_granted':False}


def futures_status():
    return {'status':'EXISTING_FUTURES_CONNECTION_NOT_VERIFIED_HERE','symbol':'/MES',
            'provider':'CME DataMine','delivery_model':'Historical API and files; no live stream asserted',
            'configuration_state':'NOT_INSPECTED','existing_configuration_changed':False,
            'adapter_enabled':None,'entitlement_verified':None,'fresh_messages_observed':None,
            'reason':'Preserve the existing futures connection. This checker has not inspected its configuration or observed its current messages; it does not require a new provider or label the existing setup unconfigured.'}


def reconcile_balances(snapshot,account):
    from infinbot_common import number
    if not isinstance(account,dict) or account.get('currency')!='USD':raise Rejected('Explicit USD broker account required; no assumed currency')
    cash_equal=number(snapshot['cash'])==number(account['cash'])
    equity_equal=number(snapshot['equity'])==number(account['equity'])
    return {'status':'BALANCE_RECONCILIATION_MATCH' if cash_equal and equity_equal else 'BALANCE_RECONCILIATION_MISMATCH',
            'cash_equal':cash_equal,'equity_equal':equity_equal,'ledger_adjusted':False,'default_funding_used':False}