main.py

"""Trade Copilot — FastAPI server.

Endpoints:
  GET /            → dashboard HTML (served from static/index.html)
  GET /api/scan    → ranked shortlist JSON
  GET /api/card/{symbol} → full trade card for one symbol
  GET /api/status  → health + active source

All computation is deterministic — no AI in the data path.
"""
from __future__ import annotations
import math, statistics, time, os, re
from pathlib import Path
from functools import lru_cache
from datetime import datetime, timezone

import asyncio
import logging
from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException, Query
from fastapi.staticfiles import StaticFiles
from fastapi.responses import JSONResponse, FileResponse

from exchange import pick_source, SOURCES, BingX, Binance
from scorer import score_symbol
from catalyst import fetch_fear_greed, fetch_coin_news, score_catalyst, fetch_crypto_headlines
from signals import scan_universe, build_signal_universe
from backtest import backtest_symbol
from signal_log import log_signal, resolve_signals, get_accuracy, get_symbol_accuracy

# CryptoPanic token — set in HF Spaces Secrets as CRYPTOPANIC_TOKEN
# Never hardcode or log this value
CRYPTOPANIC_TOKEN = os.environ.get("CRYPTOPANIC_TOKEN", "")

logger = logging.getLogger("main")

ALERT_INTERVAL = 300  # 5 minutes

# ── Synthetic-instrument hygiene (Strategy Fix Plan F5b) ───────────────────
# BingX lists non-crypto CFD proxies (oil, forex, metals, index futures) on the
# same USDT-M perpetual endpoint as real crypto. 10/50 alerts in the Jul 8-11
# batch were these — unanalyzable (no independent data source) and off-mission.
# Patterns below are derived from the 7 confirmed synthetic symbols found in
# analysis/replay_results.csv (NO_DATA rows), not guessed:
#   CO1OILBRENT2USD-USDT, NCCO1OILWTI2USD-USDT, NCCOGOLD2USD-USDT,
#   NCCOXAG2USD-USDT, NCFXGBP2JPY-USDT, NCSISP5002USD-USDT, SINASDAQ1002USD-USDT
# The 3rd pattern (digit + 3-letter fake-FX code immediately before "-USDT")
# alone catches all 7 observed cases and is low-false-positive: real crypto
# tickers (BTC-USDT, 1000PEPE-USDT, etc.) never end in <digit><3 letters>-USDT.
EXCLUDE_SYNTHETICS = os.environ.get("EXCLUDE_SYNTHETICS", "true").lower() == "true"
_SYNTHETIC_PATTERNS = [
    re.compile(r"^(CO1|NCFX|NCS|NCCO|SIN)", re.IGNORECASE),                # known prefixes
    re.compile(r"(XAU|XAG|NGAS|NATGAS|BRENT|WTI|GOLD|SILVER|NASDAQ|SP500)", re.IGNORECASE),  # commodity/index keywords
    re.compile(r"\d[A-Z]{3}-USDT$"),                                       # embedded fake-FX quote suffix
]


def is_synthetic(symbol: str) -> bool:
    return any(p.search(symbol) for p in _SYNTHETIC_PATTERNS)


# ── Alert trigger rule (confidence + PA confluence grade) ──────────────────
# Changed 13 Jul 2026 per explicit instruction: previously any card with
# confidence >= 8.8 fired an alert, no grade check. Now the confidence floor
# is lowered to 7.0, but a chart-pattern (confluence) grade of A or A+ is
# REQUIRED regardless of how high confidence is — grade B/C/D never fires,
# even at confidence 9+. Single shared gate used by both the background loop
# and manual /api/scan so the two paths can't drift apart again (that exact
# drift was the earlier website/Telegram discrepancy bug).
ALERT_CONF_MIN = 7.0
ALERT_GRADES_OK = {"A+", "A"}


def _passes_alert_gate(card: dict) -> bool:
    if card.get("vetoed", False):
        return False
    if card.get("confidence", 0) < ALERT_CONF_MIN:
        return False
    grade = (card.get("price_action") or {}).get("confluence_grade")
    return grade in ALERT_GRADES_OK


# ── Website -> Telegram alert config sync ──────────────────────────────────
# Root cause fixed here: the dashboard's account/risk/top/leverage settings
# used to live only in the browser tab and were never seen by the background
# alert loop, which hardcoded its own numbers instead. This in-memory store
# is the shared source of truth both sides now read/write. It only ever
# contains a key once the user has ACTIVELY changed that setting on the
# website — untouched keys fall back to today's existing defaults below, so
# nothing changes for anyone who never touches the settings. Resets on Space
# restart/redeploy, same as the existing dedup cache (see telegram_alert.py).
_alert_config: dict = {}
ALERT_CONFIG_BOUNDS = {
    "top": (1, 50),
    "user_leverage": (1.0, 100.0),
    "risk_pct": (0.01, 100.0),
    "account": (1.0, 10_000_000.0),
}


def _clamp(key: str, value: float) -> float:
    lo, hi = ALERT_CONFIG_BOUNDS[key]
    return max(lo, min(hi, value))


async def _alert_loop():
    """Background task: scan every 5 min and fire Telegram alerts for high-confidence cards.

    account/risk_pct/user_leverage/top scan-depth are read fresh from
    _alert_config each cycle so the website's live settings govern what the
    background loop does (fixes the website/Telegram discrepancy — previously
    this loop ignored the dashboard entirely and used fixed literals). Falls
    back to the original defaults (ALERT_ACCOUNT secret, 6x, 1%, top-30) for
    any setting the user hasn't touched yet.
    """
    from telegram_alert import send_alert, prune_sent_cache, send_raw_message
    from signal_log import log_signal
    # Fallback account size from HF Secret — used only until the website sets a custom one
    try:
        default_account = float(os.environ.get("ALERT_ACCOUNT", "10"))
    except ValueError:
        default_account = 10.0
    logger.info("Alert loop starting — default account=$%.2f (website can override live)", default_account)

    await asyncio.sleep(60)  # give app time to fully boot before first scan
    loop_count   = 0
    HEARTBEAT_EVERY = 72   # 72 × 5 min = 6 hours
    while True:
        loop_count += 1
        prune_sent_cache()  # prune stale dedup entries every cycle
        try:
            src = get_source()
            secondary = get_secondary()
            alert_account  = _alert_config.get("account", default_account)
            alert_risk_pct = _alert_config.get("risk_pct", 1.0)
            alert_leverage = _alert_config.get("user_leverage", 6.0)
            alert_top      = int(_alert_config.get("top", 30))
            candidates, _synth_excluded = rank_universe(src, src.min_vol, alert_top)
            for c in candidates:
                try:
                    card = score_symbol(src, c["symbol"], alert_account, alert_risk_pct,
                                        secondary=secondary, user_leverage=alert_leverage)
                    if _passes_alert_gate(card):
                        tv_sym = src.tv_symbol(card["symbol"])
                        sent = await send_alert(card, tv_sym)
                        # Only log alerts that actually reached Telegram (sent=True) —
                        # send_alert can still return False here on its own 4h dedup,
                        # even though this card cleared the confidence gate. Logging
                        # was previously missing entirely from this loop (root cause
                        # of "are alerts even being saved" — they weren't, anywhere).
                        if sent:
                            try:
                                log_signal(card)
                            except Exception as e:
                                logger.debug("log_signal failed for %s: %s", card.get("symbol"), e)
                        await asyncio.sleep(1)   # small gap between sends
                except Exception as e:
                    logger.debug("Alert scan error for %s: %s", c.get("symbol"), e)
                await asyncio.sleep(0.5)  # gentle rate limit between symbols

            # Resolve any logged signals past their 48h window. Cheap on most
            # cycles (resolve_signals skips anything not yet due — only fetches
            # klines for signals actually ready to check). Previously this only
            # ran if someone manually hit POST /api/resolve-signals, which
            # nothing did — logged signals would sit as "pending" forever even
            # past 48h without this.
            try:
                res = resolve_signals(src)
                if res.get("resolved"):
                    logger.info("Resolved %d signal(s): %s", res["resolved"], res)
            except Exception as e:
                logger.debug("resolve_signals failed: %s", e)
        except Exception as e:
            logger.error("Alert loop error: %s", e)

        # Heartbeat canary — every 6 h send a "system alive" message to Telegram
        if loop_count % HEARTBEAT_EVERY == 0:
            now_str = datetime.now(timezone.utc).strftime("%H:%M UTC · %d %b")
            await send_raw_message(
                f"✅ <b>Trade Copilot alive</b>\n"
                f"<i>Heartbeat #{loop_count // HEARTBEAT_EVERY} · {now_str}</i>"
            )

        await asyncio.sleep(ALERT_INTERVAL)


# Event loop reference — captured at startup so sync endpoints can schedule coroutines
_event_loop: asyncio.AbstractEventLoop | None = None


@asynccontextmanager
async def lifespan(app):
    global _event_loop
    _event_loop = asyncio.get_running_loop()
    task = asyncio.create_task(_alert_loop())
    try:
        yield
    finally:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass


app = FastAPI(title="Trade Copilot", version="1.0", lifespan=lifespan)

# Cache the active source for 5 min (avoids re-pinging on every request)
_source_cache: dict = {}
_source_ts: float = 0
SOURCE_TTL = 300

# Secondary source for cross-validation (Binance — deeper OI/funding data)
_binance: Binance | None = None
_binance_ok: bool = False
_binance_ts: float = 0
BINANCE_TTL = 300


def get_source():
    global _source_cache, _source_ts
    if time.time() - _source_ts > SOURCE_TTL or not _source_cache.get("src"):
        src = pick_source()
        _source_cache = {"src": src}
        _source_ts = time.time()
    return _source_cache["src"]


def get_secondary():
    """Returns Binance instance if reachable, else None. Used for cross-validation."""
    global _binance, _binance_ok, _binance_ts
    if time.time() - _binance_ts < BINANCE_TTL:
        return _binance if _binance_ok else None
    try:
        b = Binance()
        b.ping()
        _binance = b
        _binance_ok = True
    except Exception:
        _binance = None
        _binance_ok = False
    _binance_ts = time.time()
    return _binance if _binance_ok else None


def rank_universe(src, min_volume: float, top_n: int) -> tuple[list[dict], int]:
    """Returns (picked_candidates, synthetics_excluded_count)."""
    perps = src.universe()
    rows = []
    synth_excluded = 0
    for t in src.tickers():
        if t["symbol"] not in perps:
            continue
        if EXCLUDE_SYNTHETICS and is_synthetic(t["symbol"]):
            synth_excluded += 1
            continue
        qv = t["quote_volume_usd"]
        lo, hi = t.get("low", 0), t.get("high", 0)
        if qv < min_volume or lo <= 0:
            continue
        t["range_24h_pct"] = (hi - lo) / lo * 100
        rows.append(t)
    if synth_excluded:
        logger.info("rank_universe: %d synthetic instruments excluded", synth_excluded)
    if len(rows) < 3:
        return [{"symbol": s} for s in src.anchors], synth_excluded

    log_vols = [math.log(r["quote_volume_usd"]) for r in rows]
    ranges = [r["range_24h_pct"] for r in rows]
    mv, sv = statistics.mean(log_vols), statistics.stdev(log_vols)
    mr, sr = statistics.mean(ranges), statistics.stdev(ranges)
    for r, lv, rg in zip(rows, log_vols, ranges):
        r["scan_score"] = round(0.5 * (lv - mv) / sv + 0.5 * (rg - mr) / sr, 3)

    rows.sort(key=lambda r: r["scan_score"], reverse=True)
    picked = rows[:top_n]
    picked_syms = {r["symbol"] for r in picked}
    for sym in src.anchors:
        if sym not in picked_syms:
            extra = next((r for r in rows if r["symbol"] == sym), None)
            if extra:
                extra["anchor"] = True
                picked.append(extra)
    return picked, synth_excluded


# ── API routes ─────────────────────────────────────────────────────────────

@app.get("/api/status")
def status():
    try:
        src = get_source()
        return {"ok": True, "source": src.name, "market": src.market_note,
                "ts": datetime.now(timezone.utc).isoformat()}
    except Exception as e:
        return JSONResponse({"ok": False, "error": str(e)}, status_code=503)


@app.get("/api/alert-config")
def get_alert_config():
    """Current website->Telegram alert sync settings. is_custom=False means
    the background loop is still using its original hardcoded defaults."""
    return {
        "account": _alert_config.get("account"),
        "risk_pct": _alert_config.get("risk_pct"),
        "top": _alert_config.get("top"),
        "user_leverage": _alert_config.get("user_leverage"),
        "is_custom": bool(_alert_config),
    }


@app.post("/api/alert-config")
def set_alert_config(account: float = None, risk_pct: float = None,
                      top: int = None, user_leverage: float = None):
    """Called by the dashboard whenever the user changes a setting. The
    background alert loop picks these up on its next 5-min cycle."""
    if account is not None:
        _alert_config["account"] = _clamp("account", account)
    if risk_pct is not None:
        _alert_config["risk_pct"] = _clamp("risk_pct", risk_pct)
    if top is not None:
        _alert_config["top"] = int(_clamp("top", top))
    if user_leverage is not None:
        _alert_config["user_leverage"] = _clamp("user_leverage", user_leverage)
    logger.info("Alert config updated from website: %s", _alert_config)
    return {"ok": True, "config": _alert_config}


@app.get("/api/scan")
def scan(top: int = 8, min_volume: float = None, account: float = 100000,
         risk_pct: float = 1.0,
         user_leverage: float = Query(default=6.0, ge=1.0, le=100.0)):
    try:
        src = get_source()
        floor = min_volume or src.min_vol
        candidates, synth_excluded = rank_universe(src, floor, top)
        secondary = get_secondary()
        cards = []
        for c in candidates:
            try:
                card = score_symbol(src, c["symbol"], account, risk_pct,
                                    secondary=secondary, user_leverage=user_leverage)
                card["rank_info"] = {k: c.get(k) for k in
                                     ("scan_score","change_24h_pct","range_24h_pct",
                                      "quote_volume_usd","anchor")}
                cards.append(card)
                # Auto-log every non-vetoed signal for live accuracy tracking
                try:
                    log_signal(card)
                except Exception:
                    pass
                # Fire Telegram alert per _passes_alert_gate (confidence >= 7.0 AND PA grade A/A+)
                try:
                    if _passes_alert_gate(card) and _event_loop is not None:
                        from telegram_alert import send_alert
                        tv_sym = src.tv_symbol(card["symbol"])
                        asyncio.run_coroutine_threadsafe(
                            send_alert(card, tv_sym, manual=True), _event_loop
                        )
                except Exception:
                    pass
                time.sleep(0.05)
            except Exception as e:
                cards.append({"symbol": c["symbol"], "error": str(e)[:120]})
        cards.sort(key=lambda c: c.get("confidence", 0), reverse=True)
        return {"fetched_at": datetime.now(timezone.utc).isoformat(),
                "source": src.name, "market": src.market_note, "cards": cards,
                "synthetics_excluded": synth_excluded}
    except Exception as e:
        raise HTTPException(503, str(e))


@app.get("/api/prices")
def prices(symbols: str = ""):
    """Lightweight price ticker — returns current price + 24h change for a
    comma-separated list of symbols. Used for real-time card price updates."""
    try:
        src = get_source()
        requested = {s.strip() for s in symbols.split(",") if s.strip()}
        tickers = src.tickers()
        result = {}
        for t in tickers:
            sym = t.get("symbol", "")
            if not requested or sym in requested:
                result[sym] = {
                    "price": t.get("last_price"),
                    "change_24h_pct": t.get("change_24h_pct"),
                    "high": t.get("high"),
                    "low": t.get("low"),
                    "quote_volume_usd": t.get("quote_volume_usd"),
                }
        return {"ts": datetime.now(timezone.utc).isoformat(), "prices": result}
    except Exception as e:
        raise HTTPException(503, str(e))


@app.get("/api/card/{symbol}")
def card(symbol: str, account: float = 100000, risk_pct: float = 1.0,
         user_leverage: float = Query(default=6.0, ge=1.0, le=100.0)):
    try:
        src = get_source()
        secondary = get_secondary()
        result = score_symbol(src, symbol.upper(), account, risk_pct,
                              secondary=secondary, user_leverage=user_leverage)
        # Log the signal for live accuracy tracking
        try:
            log_signal(result)
        except Exception:
            pass
        return result
    except Exception as e:
        raise HTTPException(503, str(e))


@app.get("/api/signals")
def signals():
    """Scan mid-cap perps (volume ranks 6-50) for event-driven signals.

    Uses 5 independent detectors: RSI extreme cross, volume surge,
    BB squeeze release, funding rate extreme, and 24h price surge.
    Each fires independently — no combined gate required.
    BTC and ETH always appended as market-wide sentiment anchors.

    Results cached 5 min per symbol inside signals.py.
    """
    try:
        src = get_source()
        secondary = get_secondary()

        # Build universe: mid-cap ranks 6-50 by USD volume + BTC/ETH anchors
        try:
            symbols, ticker_map = build_signal_universe(src)
        except Exception:
            symbols = getattr(src, "anchors", [])[:20]
            ticker_map = {}

        alerts = scan_universe(src, symbols, secondary=secondary, ticker_map=ticker_map)
        return {
            "fetched_at": datetime.now(timezone.utc).isoformat(),
            "source": src.name,
            "scanned": len(symbols),
            "alerts": alerts,
        }
    except Exception as e:
        raise HTTPException(503, str(e))


@app.get("/api/news")
def news():
    """Free crypto news — CoinGecko trending + CoinDesk + Cointelegraph RSS.

    No API key required. Returns:
      trending_coins: top 7 coins by CoinGecko search volume in last 24h
      headlines:      latest headlines from CoinDesk + Cointelegraph RSS
                      each tagged with coins_mentioned[]
    Cached 10 min.
    """
    try:
        result = fetch_crypto_headlines()
        return result
    except Exception as e:
        raise HTTPException(503, str(e))


@app.get("/api/signal-accuracy")
def signal_accuracy(symbol: str = ""):
    """Live signal accuracy — real win rate from logged + resolved signals.

    This is ground truth: every card the scanner generates gets logged,
    then checked 48h later to see if TP1 or SL was hit first.

    ?symbol=BTC-USDT  (optional — returns per-symbol breakdown)
    """
    try:
        overall = get_accuracy()
        if symbol:
            overall["symbol_breakdown"] = get_symbol_accuracy(symbol.upper())
        return overall
    except Exception as e:
        raise HTTPException(503, str(e))


@app.post("/api/resolve-signals")
def resolve_signals_endpoint():
    """Trigger resolution of all pending signals that are past their 48h window.

    Called manually or by a scheduler. Checks price range on exchange
    to determine if TP1 or SL was touched during the hold period.
    """
    try:
        src = get_source()
        result = resolve_signals(src)
        return result
    except Exception as e:
        raise HTTPException(503, str(e))


@app.get("/api/catalyst")
def catalyst(symbol: str = ""):
    """Debug endpoint — returns raw Fear & Greed + CryptoPanic data.
    ?symbol=BTC-USDT  (optional; returns coin news if provided)
    """
    try:
        fng = fetch_fear_greed()
        result: dict = {
            "fear_greed": {
                "value": fng.get("value"),
                "label": fng.get("label"),
                "score_mod": fng.get("score_mod"),
                "error": fng.get("error"),
            }
        }
        if symbol:
            news = fetch_coin_news(symbol.upper())
            combined_score, notes, _raw = score_catalyst(symbol.upper())
            result["symbol"] = symbol.upper()
            result["news"] = {
                "score": news.get("score"),
                "label": news.get("label"),
                "source": news.get("source"),
                "items": news.get("items", []),
                "error": news.get("error"),
            }
            result["catalyst_score"] = combined_score
            result["notes"] = notes
        result["cryptopanic_key_set"] = bool(CRYPTOPANIC_TOKEN)
        result["ts"] = datetime.now(timezone.utc).isoformat()
        return result
    except Exception as e:
        raise HTTPException(503, str(e))


@app.get("/api/test-telegram")
def test_telegram():
    """Diagnose Telegram relay: probe Worker reachability then send message."""
    import requests as req
    worker_url = os.environ.get("CLOUDFLARE_WORKER_URL", "")
    if not worker_url:
        return JSONResponse({"ok": False, "error": "CLOUDFLARE_WORKER_URL not set in HF Secrets"}, status_code=500)

    # Probe: GET the worker (expect 405 - proves HF can reach workers.dev)
    probe = {}
    try:
        r = req.get(worker_url, timeout=8)
        probe["worker_get"] = {"status": r.status_code, "body": r.text[:80]}
    except Exception as e:
        probe["worker_get"] = {"error": type(e).__name__ + ": " + str(e)[:120]}

    # Send the Telegram message with longer timeout
    msg = "Test: Trade Copilot relay check via Cloudflare Worker."
    try:
        resp = req.post(worker_url, json={"text": msg, "parse_mode": "HTML"}, timeout=30)
        body = resp.json()
        if resp.status_code == 200:
            return {"ok": True, "message": "Test message sent via CF Worker", "probe": probe}
        return JSONResponse({"ok": False, "status": resp.status_code, "error": body, "probe": probe}, status_code=500)
    except Exception as e:
        return JSONResponse({"ok": False, "error": type(e).__name__ + ": " + str(e)[:200], "probe": probe}, status_code=500)


@app.get("/api/backtest/{symbol}")
def backtest(symbol: str):
    """Walk-forward backtest for one symbol over ~6 months of 1h candles.

    Win condition: TP2 hit before SL.
    Results cached 6h — first call is slow (~5-15s), subsequent calls instant.

    Returns:
      total_trades, win_rate, avg_r, max_drawdown_r,
      by_state {BULL/BEAR/RANGING: {trades, wins, win_rate, avg_r}},
      by_direction {long/short: ...},
      equity_curve [float list],
      recent_trades [last 50]
    """
    try:
        src = get_source()
        result = backtest_symbol(src, symbol.upper())
        return result
    except Exception as e:
        raise HTTPException(503, str(e))


# ── Static dashboard ───────────────────────────────────────────────────────

static_dir = Path(__file__).parent / "static"
if static_dir.exists():
    app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")

@app.get("/")
def root():
    index = static_dir / "index.html"
    if index.exists():
        return FileResponse(str(index))
    return {"message": "Trade Copilot API running. Use /api/scan or /api/status."}


if __name__ == "__main__":
    import uvicorn
    port = int(os.environ.get("PORT", 7860))
    uvicorn.run("main:app", host="0.0.0.0", port=port)