"""IronDome Cloud API — Fase 1: ingestão + analytics multi-tenant.

Stack: FastAPI + asyncpg + Postgres com RLS.
Auth: header `x-api-key` resolve tenant_id, RLS isola dados.
"""
import asyncio
import json
import os
import time
from collections import defaultdict
from contextlib import asynccontextmanager
from typing import Optional

import asyncpg
import httpx
from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.responses import JSONResponse, FileResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field

DB_URL = os.environ.get('DATABASE_URL', 'postgresql://irondome_app@172.28.0.1:5432/irondome')

pool: Optional[asyncpg.Pool] = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    global pool
    pool = await asyncpg.create_pool(DB_URL, min_size=2, max_size=10)
    yield
    await pool.close()


app = FastAPI(title='IronDome Cloud', version='0.1.0', lifespan=lifespan)


# ---------- SSE broadcast (in-memory por tenant) ----------
_subs: dict[int, set[asyncio.Queue]] = defaultdict(set)
_subs_lock = asyncio.Lock()


async def _broadcast(tenant_id: int, payload: dict):
    """Empurra evento pros subscribers.
    - crash: GLOBAL (todos os tenants veem, Aviator e provably-fair global)
    - bet_*: PER-TENANT (cada cliente vê só os próprios bets)"""
    is_global = payload.get("type") == "crash"
    async with _subs_lock:
        if is_global:
            all_queues = []
            for qs in _subs.values():
                all_queues.extend(qs)
            queues = all_queues
        else:
            queues = list(_subs.get(tenant_id, set()))
    for q in queues:
        try:
            q.put_nowait(payload)
        except asyncio.QueueFull:
            pass


async def _add_subscriber(tenant_id: int) -> asyncio.Queue:
    q: asyncio.Queue = asyncio.Queue(maxsize=200)
    async with _subs_lock:
        _subs[tenant_id].add(q)
    return q


async def _remove_subscriber(tenant_id: int, q: asyncio.Queue):
    async with _subs_lock:
        if tenant_id in _subs:
            _subs[tenant_id].discard(q)
            if not _subs[tenant_id]:
                del _subs[tenant_id]


# ---------- helpers ----------
async def resolve_tenant(api_key: str) -> int:
    if not api_key:
        raise HTTPException(401, 'missing x-api-key')
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT id FROM tenants WHERE api_key = $1 AND status = 'active'", api_key
        )
        if not row:
            raise HTTPException(401, 'invalid api key')
        return row['id']


async def resolve_tenant_full(api_key: str) -> dict:
    if not api_key:
        raise HTTPException(401, 'missing x-api-key')
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT id, name, role FROM tenants WHERE api_key = $1 AND status = 'active'",
            api_key,
        )
        if not row:
            raise HTTPException(401, 'invalid api key')
        return dict(row)


async def require_admin(api_key: str) -> int:
    me = await resolve_tenant_full(api_key)
    if me['role'] != 'admin':
        raise HTTPException(403, 'admin only')
    return me['id']


async def with_tenant(tenant_id: int):
    """Context manager que aplica RLS via SET LOCAL."""
    conn = await pool.acquire()
    try:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        yield conn
    finally:
        await pool.release(conn)


# ---------- models ----------
class CrashIn(BaseModel):
    ts: float
    value: float = Field(ge=1.0, lt=10000)
    inferred: bool = False


class BetIn(BaseModel):
    ts: float
    panel: int
    target: float
    amount: float
    ia_approved: bool = False
    pattern: Optional[str] = None
    probability: Optional[float] = None


class BetResolveIn(BaseModel):
    bet_id: int
    hit: bool
    delta_pnl: float
    resolved_ts: float


# ---------- endpoints ----------
@app.get('/healthz')
async def healthz():
    return {'ok': True, 'ts': time.time()}

# ---------- Telegram outbound (fase 1 - aba Configuração) ----------
async def _tg_send_if_enabled(conn, tenant_id: int, kind: str, text: str):
    """Envia mensagem Telegram se telegram_enabled=true E tg_notify_<kind>=true.
    kind: bet | hit | miss | stop | session | pause.
    Não bloqueia: dispara em background task."""
    col = f'tg_notify_{kind}'
    row = await conn.fetchrow(
        f"SELECT telegram_enabled, telegram_bot_token, telegram_chat_id, {col} AS notify FROM operator_state WHERE tenant_id=$1",
        tenant_id,
    )
    if not row: return
    if not row['telegram_enabled']: return
    if not row['notify']: return
    token = row['telegram_bot_token'] or ''
    chat = row['telegram_chat_id'] or ''
    if not token or not chat: return
    asyncio.create_task(_tg_post(token, chat, text))

async def _tg_post(token: str, chat: str, text: str):
    try:
        url = f'https://api.telegram.org/bot{token}/sendMessage'
        async with httpx.AsyncClient(timeout=10) as client:
            await client.post(url, json={'chat_id': chat, 'text': text, 'parse_mode': 'HTML'})
    except Exception as e:
        print(f'[tg] send falhou: {e}')




@app.get('/api/v1/me')
async def me(x_api_key: str = Header(default='')):
    info = await resolve_tenant_full(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(info['id']))
        flags = await conn.fetchrow("SELECT auto_click_enabled, tab_strategies_enabled, tab_analysis_enabled, tab_config_enabled, history_limit_rounds FROM tenants WHERE id=$1", info['id'])
    return {'id': info['id'], 'name': info['name'], 'role': info['role'], 'features': dict(flags) if flags else {}}


# ---------- admin: gestão de tenants ----------
class TenantCreate(BaseModel):
    name: str = Field(min_length=2, max_length=80)
    auto_click_enabled: bool = False
    tab_strategies_enabled: bool = True
    tab_analysis_enabled: bool = True
    tab_config_enabled: bool = True
    history_limit_rounds: int = Field(default=0, ge=0, le=100000)


class TenantUpdate(BaseModel):
    status: Optional[str] = Field(default=None, pattern='^(active|disabled)$')
    role: Optional[str] = Field(default=None, pattern='^(admin|user)$')
    name: Optional[str] = Field(default=None, min_length=2, max_length=80)
    auto_click_enabled: Optional[bool] = None
    tab_strategies_enabled: Optional[bool] = None
    tab_analysis_enabled: Optional[bool] = None
    tab_config_enabled: Optional[bool] = None
    history_limit_rounds: Optional[int] = Field(default=None, ge=0, le=100000)


class OperatorStateUpdate(BaseModel):
    enabled: Optional[bool] = None
    strategy: Optional[str] = Field(default=None, pattern='^(CONSERVADOR|MODERADO|AGRESSIVO)$')
    unit_brl: Optional[float] = Field(default=None, gt=0, le=10000)
    stoploss_brl: Optional[float] = Field(default=None, ge=0, le=100000)
    stopgain_brl: Optional[float] = Field(default=None, ge=0, le=100000)
    house: Optional[str] = Field(default=None, pattern='^(7K|ESTRELABET|BETANO|PIXBET)$')
    telegram_enabled: Optional[bool] = None
    telegram_bot_token: Optional[str] = Field(default=None, max_length=200)
    telegram_chat_id: Optional[str] = Field(default=None, max_length=50)
    jitter_min_ms: Optional[int] = Field(default=None, ge=0, le=10000)
    jitter_max_ms: Optional[int] = Field(default=None, ge=0, le=10000)
    heatmap_filter_pct: Optional[float] = Field(default=None, ge=0, le=100)
    heatmap_auto: Optional[bool] = None
    # Configuração avançada (Bloco 1-5 da aba Configuração)
    cap_diario_limit: Optional[int] = Field(default=None, ge=0, le=10000)
    zgap_window_days: Optional[int] = Field(default=None, ge=0, le=365)
    observe_mode: Optional[str] = Field(default=None, pattern='^(off|rounds|tail|zgap_streak)$')
    observe_n_rounds: Optional[int] = Field(default=None, ge=0, le=100)
    tail_enabled: Optional[bool] = None
    tail_window_n: Optional[int] = Field(default=None, ge=1, le=20)
    tail_prob_min: Optional[float] = Field(default=None, ge=0, le=100)
    cooldown_n_misses: Optional[int] = Field(default=None, ge=0, le=20)
    cooldown_mode: Optional[str] = Field(default=None, pattern='^(timer|shadow)$')
    cooldown_value: Optional[int] = Field(default=None, ge=1, le=10000)
    cooldown_k_min: Optional[int] = Field(default=None, ge=0, le=100)
    op_window_start: Optional[str] = Field(default=None, pattern='^[0-2][0-9]:[0-5][0-9]$')
    op_window_end: Optional[str] = Field(default=None, pattern='^[0-2][0-9]:[0-5][0-9]$')
    op_days: Optional[str] = Field(default=None, pattern='^[1-7]+$')
    auto_reset_daily: Optional[bool] = None
    tg_notify_bet: Optional[bool] = None
    tg_notify_hit: Optional[bool] = None
    tg_notify_miss: Optional[bool] = None
    tg_notify_stop: Optional[bool] = None
    tg_notify_session: Optional[bool] = None
    tg_notify_pause: Optional[bool] = None


class OperatorHeartbeat(BaseModel):
    current_round_idx: Optional[int] = None
    cycles_completed: Optional[int] = None
    cycles_lost: Optional[int] = None
    pnl_total_brl: Optional[float] = None
    daily_pnl_brl: Optional[float] = None


# Catálogo das 3 estratégias servido pra o operator (mesmo JSON do dashboard)
STRATEGIES_3 = {
  "CONSERVADOR": {"display_name": "Conservador", "rounds": 19, "schedule": [
    {"round": 1, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 3, "b2_target": 10.0},
    {"round": 2, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 3, "b2_target": 10.0},
    {"round": 3, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 3, "b2_target": 10.0},
    {"round": 4, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 3, "b2_target": 10.0},
    {"round": 5, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 3, "b2_target": 10.0},
    {"round": 6, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 3, "b2_target": 10.0},
    {"round": 7, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 3, "b2_target": 10.0},
    {"round": 8, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 4, "b2_target": 10.0},
    {"round": 9, "b1_stake": 2, "b1_target": 4.0, "b2_stake": 6, "b2_target": 10.0},
    {"round": 10, "b1_stake": 2, "b1_target": 4.0, "b2_stake": 6, "b2_target": 10.0},
    {"round": 11, "b1_stake": 2, "b1_target": 4.0, "b2_stake": 6, "b2_target": 10.0},
    {"round": 12, "b1_stake": 2, "b1_target": 4.0, "b2_stake": 7, "b2_target": 10.0},
    {"round": 13, "b1_stake": 3, "b1_target": 4.0, "b2_stake": 9, "b2_target": 10.0},
    {"round": 14, "b1_stake": 3, "b1_target": 4.0, "b2_stake": 9, "b2_target": 10.0},
    {"round": 15, "b1_stake": 3, "b1_target": 4.0, "b2_stake": 10, "b2_target": 10.0},
    {"round": 16, "b1_stake": 4, "b1_target": 4.0, "b2_stake": 12, "b2_target": 10.0},
    {"round": 17, "b1_stake": 4, "b1_target": 4.0, "b2_stake": 14, "b2_target": 10.0},
    {"round": 18, "b1_stake": 5, "b1_target": 4.0, "b2_stake": 16, "b2_target": 10.0},
    {"round": 19, "b1_stake": 7, "b1_target": 3.43, "b2_stake": 18, "b2_target": 10.0},
  ]},
  "MODERADO": {"display_name": "Moderado", "rounds": 21, "schedule": [
    {"round": 1, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 4, "b2_target": 10.0},
    {"round": 2, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 4, "b2_target": 10.0},
    {"round": 3, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 4, "b2_target": 10.0},
    {"round": 4, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 4, "b2_target": 10.0},
    {"round": 5, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 4, "b2_target": 10.0},
    {"round": 6, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 4, "b2_target": 10.0},
    {"round": 7, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 4, "b2_target": 10.0},
    {"round": 8, "b1_stake": 2, "b1_target": 4.0, "b2_stake": 7, "b2_target": 10.0},
    {"round": 9, "b1_stake": 2, "b1_target": 4.0, "b2_stake": 7, "b2_target": 10.0},
    {"round": 10, "b1_stake": 2, "b1_target": 4.0, "b2_stake": 7, "b2_target": 10.0},
    {"round": 11, "b1_stake": 2, "b1_target": 4.0, "b2_stake": 7, "b2_target": 10.0},
    {"round": 12, "b1_stake": 3, "b1_target": 4.0, "b2_stake": 10, "b2_target": 10.0},
    {"round": 13, "b1_stake": 3, "b1_target": 4.0, "b2_stake": 10, "b2_target": 10.0},
    {"round": 14, "b1_stake": 4, "b1_target": 4.0, "b2_stake": 12, "b2_target": 10.0},
    {"round": 15, "b1_stake": 4, "b1_target": 4.0, "b2_stake": 12, "b2_target": 10.0},
    {"round": 16, "b1_stake": 5, "b1_target": 4.0, "b2_stake": 15, "b2_target": 10.0},
    {"round": 17, "b1_stake": 6, "b1_target": 4.0, "b2_stake": 18, "b2_target": 10.0},
    {"round": 18, "b1_stake": 6, "b1_target": 4.0, "b2_stake": 18, "b2_target": 10.0},
    {"round": 19, "b1_stake": 8, "b1_target": 4.0, "b2_stake": 22, "b2_target": 10.0},
    {"round": 20, "b1_stake": 9, "b1_target": 4.0, "b2_stake": 25, "b2_target": 10.0},
    {"round": 21, "b1_stake": 10, "b1_target": 4.0, "b2_stake": 30, "b2_target": 10.0},
  ]},
  "AGRESSIVO": {"display_name": "Agressivo", "rounds": 21, "schedule": [
    {"round": 1, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 5, "b2_target": 10.0},
    {"round": 2, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 5, "b2_target": 10.0},
    {"round": 3, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 5, "b2_target": 10.0},
    {"round": 4, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 5, "b2_target": 10.0},
    {"round": 5, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 5, "b2_target": 10.0},
    {"round": 6, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 5, "b2_target": 10.0},
    {"round": 7, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 5, "b2_target": 10.0},
    {"round": 8, "b1_stake": 1, "b1_target": 4.0, "b2_stake": 5, "b2_target": 10.0},
    {"round": 9, "b1_stake": 3, "b1_target": 4.0, "b2_stake": 10, "b2_target": 10.0},
    {"round": 10, "b1_stake": 3, "b1_target": 4.0, "b2_stake": 10, "b2_target": 10.0},
    {"round": 11, "b1_stake": 3, "b1_target": 4.0, "b2_stake": 10, "b2_target": 10.0},
    {"round": 12, "b1_stake": 3, "b1_target": 4.0, "b2_stake": 10, "b2_target": 10.0},
    {"round": 13, "b1_stake": 5, "b1_target": 4.0, "b2_stake": 15, "b2_target": 10.0},
    {"round": 14, "b1_stake": 5, "b1_target": 4.0, "b2_stake": 15, "b2_target": 10.0},
    {"round": 15, "b1_stake": 5, "b1_target": 4.0, "b2_stake": 15, "b2_target": 10.0},
    {"round": 16, "b1_stake": 7, "b1_target": 4.0, "b2_stake": 20, "b2_target": 10.0},
    {"round": 17, "b1_stake": 7, "b1_target": 4.0, "b2_stake": 20, "b2_target": 10.0},
    {"round": 18, "b1_stake": 9, "b1_target": 4.0, "b2_stake": 25, "b2_target": 10.0},
    {"round": 19, "b1_stake": 9, "b1_target": 4.0, "b2_stake": 27, "b2_target": 10.0},
    {"round": 20, "b1_stake": 10, "b1_target": 4.0, "b2_stake": 30, "b2_target": 10.0},
    {"round": 21, "b1_stake": 12, "b1_target": 4.0, "b2_stake": 35, "b2_target": 10.0},
  ]},
}


async def _ensure_operator_row(conn, tenant_id: int):
    """Garante que existe linha em operator_state pra esse tenant."""
    await conn.execute(
        "INSERT INTO operator_state (tenant_id) VALUES ($1) ON CONFLICT DO NOTHING",
        tenant_id,
    )


async def _check_auto_click_allowed(conn, tenant_id: int):
    """Levanta 403 se o tenant não tem auto_click_enabled."""
    row = await conn.fetchrow(
        "SELECT auto_click_enabled FROM tenants WHERE id=$1",
        tenant_id,
    )
    if not row or not row['auto_click_enabled']:
        raise HTTPException(403, 'auto-click não habilitado pro seu plano')




def _gen_api_key() -> str:
    import secrets
    return 'idk_' + secrets.token_urlsafe(32)


def _gen_license_key() -> str:
    import secrets
    return 'idl_' + secrets.token_urlsafe(24)


@app.post('/api/v1/admin/tenants')
async def admin_create_tenant(payload: TenantCreate, x_api_key: str = Header(default='')):
    await require_admin(x_api_key)
    api_key = _gen_api_key()
    license_key = _gen_license_key()
    async with pool.acquire() as conn:
        try:
            row = await conn.fetchrow(
                """INSERT INTO tenants (name, license_key, api_key, status, role,
                                      auto_click_enabled, tab_strategies_enabled,
                                      tab_analysis_enabled, tab_config_enabled, history_limit_rounds)
                   VALUES ($1, $2, $3, 'active', 'user', $4, $5, $6, $7, $8)
                   RETURNING id, name, license_key, api_key, status, role, created_at""",
                payload.name, license_key, api_key,
                payload.auto_click_enabled, payload.tab_strategies_enabled,
                payload.tab_analysis_enabled, payload.tab_config_enabled,
                payload.history_limit_rounds,
            )
        except asyncpg.UniqueViolationError:
            raise HTTPException(409, 'name or key collision; try again')
    return dict(row)


def _mask_key(k):
    if not k or len(k) < 12: return '***'
    return k[:6] + '...' + k[-4:]


@app.get('/api/v1/admin/tenants')
async def admin_list_tenants(x_api_key: str = Header(default='')):
    await require_admin(x_api_key)
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            """SELECT t.id, t.name, t.license_key, t.api_key, t.status, t.role, t.created_at, t.auto_click_enabled,
                      t.tab_strategies_enabled, t.tab_analysis_enabled, t.tab_config_enabled, t.history_limit_rounds,
                      (SELECT COUNT(*) FROM crashes c WHERE c.tenant_id = t.id)::bigint AS crashes,
                      (SELECT COUNT(*) FROM bets b WHERE b.tenant_id = t.id)::bigint AS bets
               FROM tenants t
               ORDER BY t.id"""
        )
    out = []
    for r in rows:
        d = dict(r)
        # 1.27.4 SEC-FIX: nunca retorna api_key/license_key em listagem
        d['api_key'] = _mask_key(d.pop('api_key', ''))
        d['license_key'] = _mask_key(d.pop('license_key', ''))
        out.append(d)
    return {'tenants': out}


@app.get('/api/v1/admin/tenants/{tid}/reveal_key')
async def admin_reveal_key(tid: int, x_api_key: str = Header(default='')):
    """1.27.4 — endpoint separado pra ver a key COMPLETA de UM tenant específico.
    Use só quando precisar copiar pro cliente (botão na UI)."""
    await require_admin(x_api_key)
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            'SELECT api_key, license_key FROM tenants WHERE id = $1', tid
        )
    if not row:
        raise HTTPException(404, 'tenant not found')
    return {'api_key': row['api_key'], 'license_key': row['license_key']}


@app.patch('/api/v1/admin/tenants/{tid}')
async def admin_update_tenant(tid: int, payload: TenantUpdate, x_api_key: str = Header(default='')):
    me_id = await require_admin(x_api_key)
    if not any([payload.status, payload.role, payload.name, payload.auto_click_enabled is not None, payload.tab_strategies_enabled is not None, payload.tab_analysis_enabled is not None, payload.tab_config_enabled is not None, payload.history_limit_rounds is not None]):
        raise HTTPException(400, 'nothing to update')
    if tid == me_id and (payload.role == 'user' or payload.status == 'disabled'):
        raise HTTPException(400, 'cannot demote/disable yourself')
    sets, params, idx = [], [], 1
    if payload.status is not None:
        sets.append(f'status = ${idx}'); params.append(payload.status); idx += 1
    if payload.role is not None:
        sets.append(f'role = ${idx}'); params.append(payload.role); idx += 1
    if payload.name is not None:
        sets.append(f'name = ${idx}'); params.append(payload.name); idx += 1
    if payload.auto_click_enabled is not None:
        sets.append(f"auto_click_enabled = ${idx}")
        params.append(payload.auto_click_enabled)
        idx += 1
    if payload.tab_strategies_enabled is not None:
        sets.append(f"tab_strategies_enabled = ${idx}")
        params.append(payload.tab_strategies_enabled)
        idx += 1
    if payload.tab_analysis_enabled is not None:
        sets.append(f"tab_analysis_enabled = ${idx}")
        params.append(payload.tab_analysis_enabled)
        idx += 1
    if payload.tab_config_enabled is not None:
        sets.append(f"tab_config_enabled = ${idx}")
        params.append(payload.tab_config_enabled)
        idx += 1
    if payload.history_limit_rounds is not None:
        sets.append(f"history_limit_rounds = ${idx}")
        params.append(payload.history_limit_rounds)
        idx += 1
    params.append(tid)
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            f'UPDATE tenants SET ' + ', '.join(sets) +
            f' WHERE id = ${idx} RETURNING id, name, status, role',
            *params,
        )
        if not row:
            raise HTTPException(404, 'tenant not found')
    return dict(row)


@app.post('/api/v1/admin/tenants/{tid}/rotate_key')
async def admin_rotate_key(tid: int, x_api_key: str = Header(default='')):
    await require_admin(x_api_key)
    new_key = _gen_api_key()
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            'UPDATE tenants SET api_key = $1 WHERE id = $2 RETURNING id, name, api_key',
            new_key, tid,
        )
        if not row:
            raise HTTPException(404, 'tenant not found')
    return dict(row)


@app.post('/api/v1/crashes')
async def ingest_crash(payload: CrashIn, x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute(
            "SELECT set_config('app.tenant_id', $1, false)", str(tenant_id)
        )
        rec = await conn.fetchrow(
            'INSERT INTO crashes (tenant_id, ts, value, inferred) VALUES ($1, $2, $3, $4) RETURNING id',
            tenant_id, payload.ts, payload.value, payload.inferred,
        )
    await _broadcast(tenant_id, {
        'type': 'crash', 'id': rec['id'], 'ts': payload.ts, 'value': payload.value, 'inferred': payload.inferred,
    })
    return {'ok': True, 'id': rec['id']}


@app.post('/api/v1/crashes/bulk')
async def ingest_crashes_bulk(payload: list[CrashIn], x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    if not payload:
        return {'ok': True, 'inserted': 0}
    async with pool.acquire() as conn:
        await conn.execute(
            "SELECT set_config('app.tenant_id', $1, false)", str(tenant_id)
        )
        records = [(tenant_id, c.ts, c.value, c.inferred) for c in payload]
        await conn.executemany(
            'INSERT INTO crashes (tenant_id, ts, value, inferred) VALUES ($1, $2, $3, $4)', records
        )
    # broadcast só se o batch é pequeno (sync runtime). Backfill (>50 crashes)
    # é importação histórica e não precisa empurrar evento.
    if len(payload) <= 50:
        for c in payload:
            await _broadcast(tenant_id, {
                'type': 'crash', 'ts': c.ts, 'value': c.value, 'inferred': c.inferred,
            })
    return {'ok': True, 'inserted': len(payload)}


@app.post('/api/v1/bets')
async def insert_bet(payload: BetIn, x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute(
            "SELECT set_config('app.tenant_id', $1, false)", str(tenant_id)
        )
        rec = await conn.fetchrow(
            '''INSERT INTO bets (tenant_id, ts, panel, target, amount, ia_approved, pattern, probability)
               VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING id''',
            tenant_id, payload.ts, payload.panel, payload.target, payload.amount,
            payload.ia_approved, payload.pattern, payload.probability,
        )
    await _broadcast(tenant_id, {'type': 'bet_open', 'id': rec['id'], 'panel': payload.panel,
                                  'target': payload.target, 'probability': payload.probability})
    async with pool.acquire() as c2:
        await c2.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        await _tg_send_if_enabled(c2, tenant_id, 'bet',
            f"📥 <b>BET</b> P{payload.panel} @ {payload.target}× — R${payload.amount:.2f}")
    return {'ok': True, 'id': rec['id']}


@app.post('/api/v1/bets/resolve')
async def resolve_bet(payload: BetResolveIn, x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute(
            "SELECT set_config('app.tenant_id', $1, false)", str(tenant_id)
        )
        result = await conn.execute(
            '''UPDATE bets SET hit=$1, delta_pnl=$2, resolved_ts=$3
               WHERE id=$4 AND tenant_id=$5 AND resolved_ts IS NULL''',
            payload.hit, payload.delta_pnl, payload.resolved_ts,
            payload.bet_id, tenant_id,
        )
    updated = result.split()[-1] == '1' if result else False
    if updated:
        await _broadcast(tenant_id, {'type': 'bet_resolve', 'id': payload.bet_id,
                                      'hit': payload.hit, 'delta_pnl': payload.delta_pnl})
        kind = 'hit' if payload.hit else 'miss'
        emoji = '🎯' if payload.hit else '❌'
        sign = '+' if (payload.delta_pnl or 0) >= 0 else ''
        async with pool.acquire() as c2:
            await c2.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
            await _tg_send_if_enabled(c2, tenant_id, kind,
                f"{emoji} <b>{kind.upper()}</b> bet #{payload.bet_id} — {sign}R${(payload.delta_pnl or 0):.2f}")
    return {'ok': updated}


@app.get('/api/v1/stream')
async def event_stream(api_key: str, request: Request):
    """SSE: empurra eventos crash/bet em tempo real pro cliente.
    Auth via query string porque EventSource nativo não envia headers."""
    tenant_id = await resolve_tenant(api_key)
    q = await _add_subscriber(tenant_id)

    async def gen():
        try:
            yield ': connected\n\n'
            while True:
                if await request.is_disconnected():
                    break
                try:
                    ev = await asyncio.wait_for(q.get(), timeout=15.0)
                    yield f"event: {ev.get('type', 'message')}\ndata: {json.dumps(ev)}\n\n"
                except asyncio.TimeoutError:
                    # keep-alive ping pra não cair em proxies
                    yield ': ping\n\n'
        finally:
            await _remove_subscriber(tenant_id, q)

    return StreamingResponse(
        gen(),
        media_type='text/event-stream',
        headers={
            'Cache-Control': 'no-cache, no-transform',
            'X-Accel-Buffering': 'no',
            'Connection': 'keep-alive',
        },
    )


@app.get('/api/v1/crashes/recent')
async def recent_crashes(limit: int = 300, x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        hlim = await conn.fetchval('SELECT history_limit_rounds FROM tenants WHERE id=$1', tenant_id) or 0
    if hlim > 0:
        limit = max(1, min(hlim, limit))
    else:
        limit = max(1, min(2000, limit))
    async with pool.acquire() as conn:
        await conn.execute(
            "SELECT set_config('app.tenant_id', $1, false)", str(tenant_id)
        )
        rows = await conn.fetch(
            'SELECT ts, value, inferred FROM crashes ORDER BY ts DESC LIMIT $1',
            limit,
        )
    return {'crashes': [{'value': r['value'], 'at': r['ts'], 'inferred': r['inferred']} for r in reversed(rows)]}





@app.get('/api/v1/analytics/heatmap_dow_hour')
async def heatmap_dow_hour(target: float = 10.0, days: int = 0, tz: str = 'America/Sao_Paulo', x_api_key: str = Header(default='')):
    """1.28.0 — Heatmap DOW × Hora com HIT RATE real.
    DOW: 0=Seg, 1=Ter, ..., 6=Dom (no fuso passado em tz).
    days: 0 = all-time; 7/30/90 = janela das últimas N dias.
    tz: whitelist de fusos suportados."""
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        if not await conn.fetchval("SELECT tab_analysis_enabled FROM tenants WHERE id=$1", tenant_id):
            raise HTTPException(403, "feature tab_analysis_enabled disabled")
    days = max(0, int(days))
    # whitelist de timezones (proteção contra SQL injection na interpolação string)
    if tz not in ('America/Sao_Paulo', 'America/Los_Angeles'):
        tz = 'America/Sao_Paulo'
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        ts_filter = ""
        params = [float(target)]
        if days > 0:
            ts_filter = "WHERE ts >= $2"
            params.append(time.time() - days * 86400)
        rows = await conn.fetch(
            f"""SELECT
                ((EXTRACT(DOW FROM TO_TIMESTAMP(ts) AT TIME ZONE '{tz}')::int + 6) % 7) AS dow,
                EXTRACT(HOUR FROM TO_TIMESTAMP(ts) AT TIME ZONE '{tz}')::int AS hour,
                COUNT(*) AS total,
                SUM(CASE WHEN value >= $1 THEN 1 ELSE 0 END) AS hits
              FROM crashes
              {ts_filter}
              GROUP BY dow, hour
              ORDER BY dow, hour""",
            *params,
        )
    cells = []
    for r in rows:
        total = int(r['total'])
        hits = int(r['hits'])
        hit_rate = round(100.0 * hits / total, 2) if total > 0 else 0.0
        edge_pp = round(hit_rate - (100.0 / float(target)), 2)
        cells.append({'dow': int(r['dow']), 'hour': int(r['hour']),
                      'total': total, 'hits': hits,
                      'hit_rate': hit_rate, 'edge_pp': edge_pp})
    return {'cells': cells, 'target': float(target),
            'baseline_pct': round(100.0 / float(target), 2)}


@app.get('/api/v1/analytics/day_snapshots')
async def day_snapshots(day_start: int, x_api_key: str = Header(default='')):
    """1.27.0 — 96 snapshots de 15min do dia, cada um = contagem das últimas 300 rodadas
    nas faixas blue (<2x), lilac (2-10x), pink (>=10x). day_start é o unix ts da meia-noite
    local do cliente."""
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        if not await conn.fetchval("SELECT tab_analysis_enabled FROM tenants WHERE id=$1", tenant_id):
            raise HTTPException(403, "feature tab_analysis_enabled disabled")
    WINDOW = 300
    STEP = 900
    TICKS = 96
    import time as _time
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        rows = await conn.fetch(
            'SELECT ts, value FROM crashes WHERE ts >= $1 ORDER BY ts ASC',
            float(day_start - 6 * 3600),
        )
    crashes = [(float(r['ts']), float(r['value'])) for r in rows]
    snapshots = []
    now_ts = _time.time()
    for i in range(TICKS):
        tick_ts = day_start + i * STEP
        if tick_ts > now_ts:
            snapshots.append(None); continue
        lo, hi, idx = 0, len(crashes) - 1, -1
        while lo <= hi:
            m = (lo + hi) // 2
            if crashes[m][0] <= tick_ts:
                idx = m; lo = m + 1
            else:
                hi = m - 1
        if idx < 0:
            snapshots.append(None); continue
        start = max(0, idx - WINDOW + 1)
        window = crashes[start:idx + 1]
        if len(window) < 50:
            snapshots.append(None); continue
        # 1.27.1 — se o último crash da janela é mais velho que STEP+60s atrás do tick,
        # significa que houve gap de scraping; não plota (linha reta no gráfico engana).
        if (tick_ts - window[-1][0]) > (STEP + 60):
            snapshots.append(None); continue
        blue = sum(1 for _, v in window if v < 2.0)
        lilac = sum(1 for _, v in window if 2.0 <= v < 10.0)
        pink = sum(1 for _, v in window if v >= 10.0)
        snapshots.append({'i': i, 'blue': blue, 'lilac': lilac, 'pink': pink})
    return {'snapshots': snapshots, 'window': WINDOW, 'step': STEP, 'ticks': TICKS}
@app.get('/api/v1/analytics/hour-target')
async def analytics_hour_target(x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        if not await conn.fetchval("SELECT tab_analysis_enabled FROM tenants WHERE id=$1", tenant_id):
            raise HTTPException(403, "feature tab_analysis_enabled disabled")
    async with pool.acquire() as conn:
        await conn.execute(
            "SELECT set_config('app.tenant_id', $1, false)", str(tenant_id)
        )
        rows = await conn.fetch('''
            SELECT
              EXTRACT(HOUR FROM to_timestamp(ts) AT TIME ZONE 'America/Sao_Paulo')::int hora,
              t.target,
              COUNT(*) total,
              SUM(CASE WHEN value >= t.target THEN 1 ELSE 0 END) hits,
              ROUND(100.0 * SUM(CASE WHEN value >= t.target THEN 1 ELSE 0 END) / COUNT(*), 1) hit_rate
            FROM crashes
            CROSS JOIN (VALUES (2.0::float8),(3.0),(4.0),(5.0),(7.0),(10.0)) AS t(target)
            WHERE tenant_id = $1
            GROUP BY hora, t.target
            HAVING COUNT(*) >= 50
            ORDER BY hora, t.target
        ''', tenant_id)
    return {'data': [dict(r) for r in rows]}


@app.get('/api/v1/intelligence/predict')
async def predict(panel: int, target: float, x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        if not await conn.fetchval("SELECT tab_config_enabled FROM tenants WHERE id=$1", tenant_id):
            raise HTTPException(403, "feature tab_config_enabled disabled")
    if target <= 1.0 or target > 1000:
        raise HTTPException(400, 'target out of range')
    if panel not in (0, 1):
        raise HTTPException(400, 'panel must be 0 or 1')

    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        rows = await conn.fetch(
            'SELECT ts, value FROM crashes ORDER BY ts DESC LIMIT 5000',
        )

    if len(rows) < 30:
        return {
            'panel': panel, 'target': target, 'probability': 0.0, 'recommend': False,
            'reason': 'amostra insuficiente', 'sample_size': len(rows),
        }

    from datetime import datetime, timezone, timedelta
    sao = timezone(timedelta(hours=-3))

    # ordenar do mais antigo pro mais recente
    seq = [(float(r['ts']), float(r['value'])) for r in reversed(rows)]
    values = [v for _, v in seq]
    binary = ['H' if v >= target else 'L' for v in values]

    # base rate sobre últimas 200
    last_200 = values[-200:]
    base_hits = sum(1 for v in last_200 if v >= target)
    base_rate = base_hits / len(last_200)

    # hora atual de São Paulo
    current_hour = datetime.now(sao).hour
    hour_vals = [v for ts, v in seq if datetime.fromtimestamp(ts, tz=sao).hour == current_hour]
    if len(hour_vals) >= 20:
        hour_rate = sum(1 for v in hour_vals if v >= target) / len(hour_vals)
    else:
        hour_rate = base_rate

    # cauda: últimos 5 crashes encodificados como L/H
    tail_len = 5
    tail = values[-tail_len:]
    tail_pattern = ''.join(binary[-tail_len:])

    # busca de padrão na história (excluindo a cauda atual)
    pattern_hits = 0
    pattern_count = 0
    for i in range(len(binary) - tail_len * 2):
        if ''.join(binary[i:i + tail_len]) == tail_pattern and i + tail_len < len(binary):
            pattern_count += 1
            if binary[i + tail_len] == 'H':
                pattern_hits += 1

    # smoothing bayesiano: prior = base_rate, peso α
    alpha = 6
    if pattern_count > 0:
        pattern_rate = (pattern_hits + alpha * base_rate) / (pattern_count + alpha)
    else:
        pattern_rate = base_rate

    # blend final: 30% base, 25% hora, 45% padrão
    probability = base_rate * 0.30 + hour_rate * 0.25 + pattern_rate * 0.45

    break_even = 1.0 / target
    margin = 1.10  # 10% acima de break-even
    recommend = probability > break_even * margin
    edge_pct = (probability - break_even) * 100

    return {
        'panel': panel,
        'target': target,
        'probability': round(probability, 3),
        'recommend': recommend,
        'break_even': round(break_even, 3),
        'edge_pct': round(edge_pct, 1),
        'context': {
            'tail': tail,
            'tail_encoded': tail_pattern,
            'base_rate': round(base_rate, 3),
            'hour_rate': round(hour_rate, 3),
            'pattern_rate': round(pattern_rate, 3),
            'pattern_count': pattern_count,
            'pattern_hits': pattern_hits,
            'sample_size': len(values),
        },
    }


@app.get('/api/v1/intelligence/predict_all')
async def predict_all(targets: str = '2,3,4,5,10,20,50,100',
                      x_api_key: str = Header(default='')):
    """Calcula probabilidade da IA pra vários targets de uma vez."""
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        if not await conn.fetchval("SELECT tab_config_enabled FROM tenants WHERE id=$1", tenant_id):
            raise HTTPException(403, "feature tab_config_enabled disabled")
    try:
        target_list = [float(t.strip()) for t in targets.split(',') if t.strip()]
    except ValueError:
        raise HTTPException(400, 'targets inválido')
    target_list = [t for t in target_list if 1.0 < t <= 1000][:20]

    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        rows = await conn.fetch(
            'SELECT ts, value FROM crashes ORDER BY ts DESC LIMIT 5000',
        )

    if len(rows) < 30:
        return {'sample_size': len(rows), 'targets': []}

    from datetime import datetime, timezone, timedelta
    sao = timezone(timedelta(hours=-3))

    seq = [(float(r['ts']), float(r['value'])) for r in reversed(rows)]
    values = [v for _, v in seq]
    last_200 = values[-200:]
    current_hour = datetime.now(sao).hour
    hour_vals = [v for ts, v in seq if datetime.fromtimestamp(ts, tz=sao).hour == current_hour]
    tail_len = 5

    out = []
    for T in target_list:
        binary = ['H' if v >= T else 'L' for v in values]
        base_hits = sum(1 for v in last_200 if v >= T)
        base_rate = base_hits / len(last_200)
        if len(hour_vals) >= 20:
            hour_rate = sum(1 for v in hour_vals if v >= T) / len(hour_vals)
        else:
            hour_rate = base_rate
        tail_pattern = ''.join(binary[-tail_len:])
        pattern_hits = 0
        pattern_count = 0
        for i in range(len(binary) - tail_len * 2):
            if ''.join(binary[i:i + tail_len]) == tail_pattern and i + tail_len < len(binary):
                pattern_count += 1
                if binary[i + tail_len] == 'H':
                    pattern_hits += 1
        alpha = 6
        if pattern_count > 0:
            pattern_rate = (pattern_hits + alpha * base_rate) / (pattern_count + alpha)
        else:
            pattern_rate = base_rate
        probability = base_rate * 0.30 + hour_rate * 0.25 + pattern_rate * 0.45
        break_even = 1.0 / T
        edge_pct = (probability - break_even) * 100
        out.append({
            'target': T,
            'probability': round(probability, 4),
            'break_even': round(break_even, 4),
            'edge_pct': round(edge_pct, 1),
            'pattern_count': pattern_count,
            'pattern_hits': pattern_hits,
            'tail_pattern': tail_pattern,
        })
    return {'sample_size': len(values), 'targets': out}


@app.get('/api/v1/panels/stats')
async def panels_stats(since: float = 0, x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        rows = await conn.fetch('''
            SELECT panel,
                   SUM(CASE WHEN hit THEN 1 ELSE 0 END)::int wins,
                   SUM(CASE WHEN NOT hit THEN 1 ELSE 0 END)::int losses,
                   SUM(CASE WHEN ia_approved THEN 1 ELSE 0 END)::int ia_approves,
                   SUM(CASE WHEN ia_approved AND hit THEN 1 ELSE 0 END)::int ia_wins,
                   SUM(CASE WHEN ia_approved AND NOT hit THEN 1 ELSE 0 END)::int ia_losses,
                   COALESCE(SUM(delta_pnl), 0)::float pnl
            FROM bets
            WHERE tenant_id=$1 AND resolved_ts IS NOT NULL AND amount > 0 AND ts >= $2
            GROUP BY panel
        ''', tenant_id, float(since))
    base = {p: {'wins': 0, 'losses': 0, 'ia_approves': 0, 'ia_wins': 0, 'ia_losses': 0, 'pnl': 0.0} for p in (0, 1)}
    for r in rows:
        p = int(r['panel'])
        if p in base:
            base[p] = {k: r[k] for k in ('wins', 'losses', 'ia_approves', 'ia_wins', 'ia_losses', 'pnl')}
    return {'panels': base}


@app.get('/api/v1/analytics/ia')
async def analytics_ia(target: Optional[float] = None, tol: float = 0.25,
                       x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    if target is not None and target <= 1.0:
        raise HTTPException(400, 'target must be > 1')
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        if target is None:
            rows = await conn.fetch('''
                WITH buckets AS (
                    SELECT
                        CASE
                            WHEN probability < 0.50 THEN '<50%'
                            WHEN probability < 0.60 THEN '50-60%'
                            WHEN probability < 0.70 THEN '60-70%'
                            WHEN probability < 0.80 THEN '70-80%'
                            WHEN probability < 0.90 THEN '80-90%'
                            ELSE '90%+'
                        END AS band,
                        hit::int hit, delta_pnl, probability
                    FROM bets
                    WHERE tenant_id=$1 AND ia_approved=true AND resolved_ts IS NOT NULL
                      AND amount > 0 AND probability IS NOT NULL
                )
                SELECT band,
                       COUNT(*)::int total,
                       SUM(hit)::int hits,
                       ROUND((100.0 * SUM(hit) / NULLIF(COUNT(*), 0))::numeric, 1)::float hit_rate,
                       ROUND((AVG(probability) * 100)::numeric, 1)::float avg_prob,
                       COALESCE(SUM(delta_pnl), 0)::float pnl
                FROM buckets
                GROUP BY band
                ORDER BY MIN(probability)
            ''', tenant_id)
        else:
            rows = await conn.fetch('''
                WITH buckets AS (
                    SELECT
                        CASE
                            WHEN probability < 0.50 THEN '<50%'
                            WHEN probability < 0.60 THEN '50-60%'
                            WHEN probability < 0.70 THEN '60-70%'
                            WHEN probability < 0.80 THEN '70-80%'
                            WHEN probability < 0.90 THEN '80-90%'
                            ELSE '90%+'
                        END AS band,
                        hit::int hit, delta_pnl, probability
                    FROM bets
                    WHERE tenant_id=$1 AND ia_approved=true AND resolved_ts IS NOT NULL
                      AND amount > 0 AND probability IS NOT NULL
                      AND target BETWEEN $2 AND $3
                )
                SELECT band,
                       COUNT(*)::int total,
                       SUM(hit)::int hits,
                       ROUND((100.0 * SUM(hit) / NULLIF(COUNT(*), 0))::numeric, 1)::float hit_rate,
                       ROUND((AVG(probability) * 100)::numeric, 1)::float avg_prob,
                       COALESCE(SUM(delta_pnl), 0)::float pnl
                FROM buckets
                GROUP BY band
                ORDER BY MIN(probability)
            ''', tenant_id, target - tol, target + tol)
    return {'bands': [dict(r) for r in rows], 'target': target, 'tol': tol if target else None}


@app.get('/api/v1/totals')
async def totals(x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute(
            "SELECT set_config('app.tenant_id', $1, false)", str(tenant_id)
        )
        c = await conn.fetchval('SELECT COUNT(*) FROM crashes WHERE tenant_id=$1', tenant_id)
        b = await conn.fetchval('SELECT COUNT(*) FROM bets WHERE tenant_id=$1', tenant_id)
    return {'crashes': c or 0, 'bets': b or 0}


# Static dashboard
app.mount('/static', StaticFiles(directory='/app/public'), name='static')






async def compute_zgap_color(conn, tenant_id: int) -> dict:
    """Porta exata do renderZGap do dashboard. Devolve {color, z}.
    color: 'green' | 'yellow' | 'red' | 'off' (sem dados suficientes)."""
    rows = await conn.fetch(
        'SELECT ts, value FROM crashes WHERE inferred=false '
        'ORDER BY ts DESC LIMIT 300',
    )
    arr = list(reversed(rows))  # mais antigo primeiro
    if len(arr) < 30:
        return {'color': 'off', 'z': None}
    intervalo_atual = 0
    for i in range(len(arr) - 1, -1, -1):
        if float(arr[i]['value']) >= 10:
            break
        intervalo_atual += 1
    hits = [i for i, r in enumerate(arr) if float(r['value']) >= 10]
    if len(hits) < 5:
        return {'color': 'off', 'z': None}
    gaps = [hits[k] - hits[k - 1] for k in range(1, len(hits))]
    avg = sum(gaps) / len(gaps)
    variance = sum((g - avg) ** 2 for g in gaps) / len(gaps)
    std = variance ** 0.5
    if std < 0.1:
        return {'color': 'off', 'z': None}
    z = (intervalo_atual - avg) / std
    if -0.5 <= z <= 0.5:
        color = 'green'
    elif z > 2.0 or z < -1.5:
        color = 'red'
    else:
        color = 'yellow'
    return {'color': color, 'z': round(z, 2)}


# ─────────── operator endpoints (1.27) ───────────

@app.get('/api/v1/strategies')
async def list_strategies(x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        if not await conn.fetchval("SELECT tab_strategies_enabled FROM tenants WHERE id=$1", tenant_id):
            raise HTTPException(403, "feature tab_strategies_enabled disabled")
    return {'strategies': [
        {'code': k, **v} for k, v in STRATEGIES_3.items()
    ]}


@app.get('/api/v1/zgap/state')
async def get_zgap_state(x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        return await compute_zgap_color(conn, tenant_id)


@app.get('/api/v1/operator/state')
async def operator_get_state(x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        await _check_auto_click_allowed(conn, tenant_id)
        await _ensure_operator_row(conn, tenant_id)
        row = await conn.fetchrow(
            "SELECT * FROM operator_state WHERE tenant_id=$1",
            tenant_id,
        )
    return dict(row) if row else {}


@app.post('/api/v1/operator/state')
async def operator_set_state(payload: OperatorStateUpdate, x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        await _check_auto_click_allowed(conn, tenant_id)
        await _ensure_operator_row(conn, tenant_id)
        # Build dynamic UPDATE
        fields = []
        values = []
        i = 1
        for field, val in payload.model_dump(exclude_unset=True).items():
            fields.append(f"{field}=${i}")
            values.append(val)
            i += 1
        if not fields:
            return {'ok': True, 'updated': 0}
        values.append(tenant_id)
        sql = f"UPDATE operator_state SET {', '.join(fields)}, updated_at=now() WHERE tenant_id=${i}"
        await conn.execute(sql, *values)
        # se acabou de habilitar e estava pausado por cap, NÃO re-habilita (precisa /unlock)
        if payload.enabled and payload.enabled is True:
            cur = await conn.fetchrow(
                "SELECT paused_reason FROM operator_state WHERE tenant_id=$1",
                tenant_id,
            )
            if cur and cur['paused_reason'] == 'cap_diario':
                await conn.execute(
                    "UPDATE operator_state SET enabled=false WHERE tenant_id=$1",
                    tenant_id,
                )
                raise HTTPException(409, 'cap_diario_atingido — use /unlock')
    return {'ok': True}


@app.post('/api/v1/operator/heartbeat')
async def operator_heartbeat(payload: OperatorHeartbeat, x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        await _check_auto_click_allowed(conn, tenant_id)
        await _ensure_operator_row(conn, tenant_id)
        # update last_heartbeat + métricas reportadas
        sets = ["last_heartbeat=now()"]
        values = []
        i = 1
        for field, val in payload.model_dump(exclude_unset=True).items():
            sets.append(f"{field}=${i}")
            values.append(val)
            i += 1
        # cap enforcement: se daily_pnl < -cap, pausa.
        # grace period: NÃO pausa se reset foi nos últimos 30s (operator pode estar
        # heartbeating com pnl velho da memória local antes de resetar)
        cur = await conn.fetchrow(
            """SELECT stoploss_brl, stopgain_brl, daily_pnl_brl, pnl_anchor, paused_reason, updated_at,
                   EXTRACT(EPOCH FROM (now() - updated_at)) AS sec_since_update
               FROM operator_state WHERE tenant_id=$1""",
            tenant_id,
        )
        new_pnl = payload.daily_pnl_brl if payload.daily_pnl_brl is not None else float(cur['daily_pnl_brl'] or 0)
        stoploss = float(cur['stoploss_brl'] or 0)
        anchor = float(cur['pnl_anchor'] or 0)
        relative_pnl = new_pnl - anchor
        stopgain = float(cur['stopgain_brl'] or 0)
        sec_since_update = float(cur['sec_since_update'] or 999)
        # Stoploss: pausa se pnl <= -stoploss (e stoploss > 0 = ativo)
        if stoploss > 0 and relative_pnl <= -stoploss and cur['paused_reason'] not in ('stoploss', 'stopgain', 'cap_diario') and sec_since_update > 30:
            sets.append("paused_reason='stoploss'")
            sets.append("paused_at=now()")
            sets.append("enabled=false")
            await _tg_send_if_enabled(conn, tenant_id, 'stop',
                f"🔒 <b>STOPLOSS</b> disparado — P&L R${relative_pnl:.2f} ≤ -R${stoploss:.2f}. Operator pausado.")
        # StopGain: pausa se pnl >= +stopgain (e stopgain > 0 = ativo)
        elif stopgain > 0 and relative_pnl >= stopgain and cur['paused_reason'] not in ('stoploss', 'stopgain', 'cap_diario') and sec_since_update > 30:
            sets.append("paused_reason='stopgain'")
            sets.append("paused_at=now()")
            sets.append("enabled=false")
            await _tg_send_if_enabled(conn, tenant_id, 'stop',
                f"✓ <b>STOPGAIN</b> atingido — P&L +R${relative_pnl:.2f} ≥ R${stopgain:.2f}. Operator pausado.")
        # Anti-flap: nos 30s pós-update, ignora pnl velho do operator (igual antes)
        if sec_since_update <= 30 and payload.daily_pnl_brl is not None and payload.daily_pnl_brl < 0 and cur['paused_reason'] is None:
            sets = [x for x in sets if not x.startswith("daily_pnl_brl=")]
            sets.append("daily_pnl_brl=0")
        values.append(tenant_id)
        sql = f"UPDATE operator_state SET {', '.join(sets)} WHERE tenant_id=${i}"
        await conn.execute(sql, *values)
        # devolve estado atualizado pro operator decidir
        row = await conn.fetchrow(
            "SELECT enabled, strategy, unit_brl, stoploss_brl, stopgain_brl, house, paused_reason FROM operator_state WHERE tenant_id=$1",
            tenant_id,
        )
        zgap = await compute_zgap_color(conn, tenant_id)
    return {'ok': True, 'state': dict(row) if row else {}, 'zgap': zgap}


@app.post('/api/v1/operator/unlock')
async def operator_unlock(x_api_key: str = Header(default='')):
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        await _check_auto_click_allowed(conn, tenant_id)
        await _ensure_operator_row(conn, tenant_id)
        await conn.execute("""
            UPDATE operator_state
            SET paused_reason=NULL, paused_at=NULL,
                pnl_anchor=daily_pnl_brl,
                updated_at=now()
            WHERE tenant_id=$1
        """, tenant_id)
    return {'ok': True}


@app.post('/api/v1/operator/reset')
async def operator_reset(x_api_key: str = Header(default='')):
    """Zera estado da sessão: P&L, round, ciclos, paused. Operator detecta
    via state_poll_loop e zera engine local (espera bet pendente resolver)."""
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        await _check_auto_click_allowed(conn, tenant_id)
        await _ensure_operator_row(conn, tenant_id)
        await conn.execute("""
            UPDATE operator_state
            SET daily_pnl_brl=0,
                pnl_anchor=0,
                current_round_idx=0,
                cycles_completed=0,
                cycles_lost=0,
                cap_diario_count=0,
                paused_reason=NULL,
                paused_at=NULL,
                updated_at=now()
            WHERE tenant_id=$1
        """, tenant_id)
        await _tg_send_if_enabled(conn, tenant_id, 'session',
            "🔄 <b>NOVA SESSÃO</b> iniciada — P&L, ciclos e contadores zerados.")
    return {'ok': True}




class TelegramTestIn(BaseModel):
    token: Optional[str] = Field(default=None, max_length=200)
    chat_id: Optional[str] = Field(default=None, max_length=50)


@app.post('/api/v1/operator/telegram_test')
async def telegram_test(payload: TelegramTestIn, x_api_key: str = Header(default='')):
    """Envia mensagem de teste pra validar credenciais Telegram.
    Usa token/chat do body se passados, senão do DB."""
    tenant_id = await resolve_tenant(x_api_key)
    async with pool.acquire() as conn:
        await conn.execute("SELECT set_config('app.tenant_id', $1, false)", str(tenant_id))
        row = await conn.fetchrow(
            "SELECT telegram_bot_token, telegram_chat_id FROM operator_state WHERE tenant_id=$1",
            tenant_id,
        )
    token = (payload.token or (row['telegram_bot_token'] if row else None) or '').strip()
    chat = (payload.chat_id or (row['telegram_chat_id'] if row else None) or '').strip()
    if not token or not chat:
        return {'ok': False, 'error': 'Bot token ou Chat ID ausentes. Preencha os campos antes.'}
    text = ('🧪 <b>Teste de notificação · IronDome</b>\n\n'
            'Sua conta está conectada. A partir de agora você vai receber '
            'os eventos configurados (BET, HIT, MISS, Stop, etc).')
    try:
        async with httpx.AsyncClient(timeout=10) as client:
            r = await client.post(
                f'https://api.telegram.org/bot{token}/sendMessage',
                json={'chat_id': chat, 'text': text, 'parse_mode': 'HTML'},
            )
        ok = r.status_code == 200
        try:
            j = r.json()
        except Exception:
            j = {}
        return {
            'ok': ok,
            'status_code': r.status_code,
            'description': j.get('description') if not ok else 'mensagem enviada',
        }
    except Exception as e:
        return {'ok': False, 'error': str(e)}


@app.get('/')
async def root():
    return FileResponse('/app/public/index.html')
