app.py

"""
MEMBRA Alpha Hunter — single-file Streamlit app.py
Self-learning prediction memory engine for perpetual futures research.

Run locally:
  streamlit run app.py

Hugging Face Space:
  SDK: Streamlit
  App file: app.py

Required packages usually needed in requirements.txt on HF Spaces:
  streamlit
  pandas
  numpy
  requests

This app is an informational research tool only. It does not place trades,
manage money, or guarantee returns.
"""

from __future__ import annotations

import csv
import hashlib
import io
import json
import math
import os
import sqlite3
import time
from dataclasses import dataclass
from datetime import datetime, timezone, timedelta
from typing import Any, Dict, Iterable, List, Optional, Tuple

import numpy as np
import pandas as pd
import requests
import streamlit as st

# =============================================================================
# Global configuration
# =============================================================================

APP_NAME = "MEMBRA Alpha Hunter"
APP_SUBTITLE = "Self-Learning Prediction Memory Engine"
DB_PATH = os.getenv("MEMBRA_DB_PATH", "membra_alpha_memory.db")
BINANCE_FAPI = "https://fapi.binance.com"
DEFAULT_MODEL = "alpha-memory-v1"
DEFAULT_SYMBOLS = "BTCUSDT,ETHUSDT,SOLUSDT,BNBUSDT,XRPUSDT,DOGEUSDT,PEPEUSDT,BSBUSDT,BILLUSDT,EDENUSDT,SKYAIUSDT,ONDOUSDT"
UTC = timezone.utc

DISCLAIMER = (
    "Informational research tool only. Not financial advice. Predictions are "
    "probabilistic and may be wrong. This app does not execute trades, custody "
    "funds, or guarantee profit. Memory learns only from closed verified outcomes."
)

# =============================================================================
# Streamlit page + styling
# =============================================================================

st.set_page_config(
    page_title=APP_NAME,
    page_icon="🧠",
    layout="wide",
    initial_sidebar_state="expanded",
)

st.markdown(
    """
<style>
:root {
  --bg: #0b0f17;
  --panel: #111827;
  --panel2: #161f2f;
  --line: #273245;
  --text: #e6edf7;
  --muted: #8b9bb0;
  --gold: #f5b942;
  --green: #1fd17c;
  --red: #ff4d6d;
  --blue: #4f8cff;
  --cyan: #3de4ff;
  --warn: #ffcc33;
}
html, body, [data-testid="stAppViewContainer"] {
  background: var(--bg);
  color: var(--text);
}
[data-testid="stSidebar"] {
  background: #090d14;
  border-right: 1px solid var(--line);
}
.main .block-container {
  padding-top: 1.3rem;
  padding-bottom: 3rem;
}
h1, h2, h3 { letter-spacing: -0.03em; }
.small-muted { color: var(--muted); font-size: 0.85rem; }
.metric-card {
  background: linear-gradient(180deg, #151f30 0%, #101722 100%);
  border: 1px solid var(--line);
  border-radius: 16px;
  padding: 15px 16px;
  box-shadow: 0 8px 24px rgba(0,0,0,0.25);
  min-height: 105px;
}
.metric-card .label { color: var(--muted); font-size: 0.78rem; text-transform: uppercase; letter-spacing: .08em; }
.metric-card .value { font-size: 1.6rem; font-weight: 800; margin-top: .35rem; }
.metric-card .sub { color: var(--muted); font-size: .78rem; margin-top: .2rem; }
.badge {
  display: inline-block;
  padding: 0.16rem 0.48rem;
  border-radius: 999px;
  font-size: 0.72rem;
  font-weight: 800;
  border: 1px solid rgba(255,255,255,0.1);
  letter-spacing: .02em;
}
.badge.green { background: rgba(31,209,124,.12); color: var(--green); }
.badge.red { background: rgba(255,77,109,.12); color: var(--red); }
.badge.blue { background: rgba(79,140,255,.13); color: var(--blue); }
.badge.gold { background: rgba(245,185,66,.14); color: var(--gold); }
.badge.gray { background: rgba(139,155,176,.13); color: var(--muted); }
.badge.cyan { background: rgba(61,228,255,.12); color: var(--cyan); }
.badge.warn { background: rgba(255,204,51,.12); color: var(--warn); }
.table-wrap {
  border: 1px solid var(--line);
  border-radius: 14px;
  overflow-x: auto;
  background: var(--panel);
}
.stDataFrame { border-radius: 14px; }
hr { border-color: var(--line) !important; }
button[kind="primary"] { font-weight: 800; }
.notice {
  border: 1px solid var(--line);
  border-left: 4px solid var(--gold);
  border-radius: 12px;
  padding: 12px 14px;
  background: rgba(245,185,66,.08);
  color: #f4e8c6;
}
</style>
""",
    unsafe_allow_html=True,
)

# =============================================================================
# Utilities
# =============================================================================


def now_utc() -> datetime:
    return datetime.now(tz=UTC)


def iso_now() -> str:
    return now_utc().isoformat(timespec="seconds")


def parse_dt(x: Any) -> Optional[datetime]:
    if not x:
        return None
    try:
        s = str(x).replace("Z", "+00:00")
        dt = datetime.fromisoformat(s)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=UTC)
        return dt.astimezone(UTC)
    except Exception:
        return None


def pct(x: Any, digits: int = 2) -> str:
    try:
        if x is None or (isinstance(x, float) and math.isnan(x)):
            return "—"
        return f"{float(x):+.{digits}f}%"
    except Exception:
        return "—"


def pct_plain(x: Any, digits: int = 1) -> str:
    try:
        if x is None or (isinstance(x, float) and math.isnan(x)):
            return "—"
        return f"{float(x):.{digits}f}%"
    except Exception:
        return "—"


def clamp(x: float, lo: float, hi: float) -> float:
    return max(lo, min(hi, x))


def stable_id(*parts: Any) -> str:
    raw = "|".join(str(p) for p in parts)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]


def normalize_symbol(s: str) -> str:
    return s.strip().upper().replace("/", "").replace("_", "")


def display_symbol(s: str) -> str:
    s = normalize_symbol(s)
    if s.endswith("USDT") and len(s) > 4:
        return s[:-4] + "/USDT"
    return s


def safe_float(x: Any, default: float = 0.0) -> float:
    try:
        if x is None:
            return default
        y = float(x)
        if math.isnan(y) or math.isinf(y):
            return default
        return y
    except Exception:
        return default


def to_bool_int(x: Any) -> int:
    return 1 if bool(x) else 0

# =============================================================================
# Database layer
# =============================================================================

SCHEMA = """
CREATE TABLE IF NOT EXISTS predictions (
    id TEXT PRIMARY KEY,
    created_at TEXT NOT NULL,
    closed_at TEXT,
    symbol TEXT NOT NULL,
    direction TEXT NOT NULL,
    setup TEXT NOT NULL,
    raw_confidence REAL NOT NULL,
    optimized_confidence REAL NOT NULL,
    alpha_score REAL NOT NULL,
    alpha_grade TEXT NOT NULL,
    action_label TEXT NOT NULL,
    model_used TEXT NOT NULL,
    rationale TEXT,
    optimizer_notes TEXT,
    entry_price REAL,
    close_price REAL,
    status TEXT NOT NULL,
    hit INTEGER,
    return_pct REAL,
    horizon_minutes INTEGER,
    verification_status TEXT NOT NULL,
    price_source TEXT,
    memory_score_at_open REAL,
    optimization_adjustment REAL,
    duplicate_flag INTEGER DEFAULT 0,
    stale_flag INTEGER DEFAULT 0,
    raw_payload TEXT
);

CREATE INDEX IF NOT EXISTS idx_predictions_symbol ON predictions(symbol);
CREATE INDEX IF NOT EXISTS idx_predictions_status ON predictions(status);
CREATE INDEX IF NOT EXISTS idx_predictions_created_at ON predictions(created_at);
CREATE INDEX IF NOT EXISTS idx_predictions_key ON predictions(symbol, direction, setup, model_used);
"""


def conn() -> sqlite3.Connection:
    c = sqlite3.connect(DB_PATH, check_same_thread=False)
    c.row_factory = sqlite3.Row
    return c


def init_db() -> None:
    with conn() as c:
        c.executescript(SCHEMA)


def insert_prediction(row: Dict[str, Any]) -> None:
    keys = [
        "id", "created_at", "closed_at", "symbol", "direction", "setup",
        "raw_confidence", "optimized_confidence", "alpha_score", "alpha_grade",
        "action_label", "model_used", "rationale", "optimizer_notes", "entry_price",
        "close_price", "status", "hit", "return_pct", "horizon_minutes",
        "verification_status", "price_source", "memory_score_at_open",
        "optimization_adjustment", "duplicate_flag", "stale_flag", "raw_payload"
    ]
    values = [row.get(k) for k in keys]
    q = f"INSERT OR REPLACE INTO predictions ({','.join(keys)}) VALUES ({','.join(['?']*len(keys))})"
    with conn() as c:
        c.execute(q, values)
        c.commit()


def update_prediction_close(
    pred_id: str,
    closed_at: str,
    close_price: float,
    hit: Optional[bool],
    return_pct: Optional[float],
    verification_status: str,
    price_source: str,
) -> None:
    with conn() as c:
        c.execute(
            """
            UPDATE predictions
            SET closed_at=?, close_price=?, status='CLOSED', hit=?, return_pct=?,
                verification_status=?, price_source=?
            WHERE id=?
            """,
            (closed_at, close_price, None if hit is None else to_bool_int(hit), return_pct, verification_status, price_source, pred_id),
        )
        c.commit()


def load_predictions(where: str = "", params: Tuple[Any, ...] = ()) -> pd.DataFrame:
    q = "SELECT * FROM predictions"
    if where:
        q += " WHERE " + where
    q += " ORDER BY created_at DESC"
    with conn() as c:
        df = pd.read_sql_query(q, c, params=params)
    return df


def count_predictions() -> Dict[str, int]:
    with conn() as c:
        rows = c.execute(
            """
            SELECT status, COUNT(*) AS n
            FROM predictions
            GROUP BY status
            """
        ).fetchall()
    out = {"OPEN": 0, "CLOSED": 0, "TOTAL": 0}
    for r in rows:
        out[str(r["status"])] = int(r["n"])
        out["TOTAL"] += int(r["n"])
    return out


def delete_all_data() -> None:
    with conn() as c:
        c.execute("DELETE FROM predictions")
        c.commit()

init_db()

# =============================================================================
# Market data
# =============================================================================


def http_get_json(url: str, params: Optional[Dict[str, Any]] = None, timeout: int = 8) -> Any:
    r = requests.get(url, params=params, timeout=timeout, headers={"User-Agent": "MembraAlphaHunter/1.0"})
    r.raise_for_status()
    return r.json()


@st.cache_data(ttl=180, show_spinner=False)
def fetch_exchange_symbols() -> List[str]:
    data = http_get_json(f"{BINANCE_FAPI}/fapi/v1/exchangeInfo")
    symbols = []
    for item in data.get("symbols", []):
        if item.get("contractType") == "PERPETUAL" and item.get("quoteAsset") == "USDT" and item.get("status") == "TRADING":
            symbols.append(item["symbol"])
    return sorted(symbols)


@st.cache_data(ttl=90, show_spinner=False)
def fetch_24h_tickers() -> pd.DataFrame:
    data = http_get_json(f"{BINANCE_FAPI}/fapi/v1/ticker/24hr")
    df = pd.DataFrame(data)
    if df.empty:
        return df
    df["quoteVolume"] = pd.to_numeric(df.get("quoteVolume"), errors="coerce").fillna(0)
    df["lastPrice"] = pd.to_numeric(df.get("lastPrice"), errors="coerce")
    df["priceChangePercent"] = pd.to_numeric(df.get("priceChangePercent"), errors="coerce")
    return df


@st.cache_data(ttl=45, show_spinner=False)
def fetch_klines(symbol: str, interval: str = "5m", limit: int = 180) -> pd.DataFrame:
    symbol = normalize_symbol(symbol)
    data = http_get_json(
        f"{BINANCE_FAPI}/fapi/v1/klines",
        params={"symbol": symbol, "interval": interval, "limit": int(limit)},
        timeout=10,
    )
    cols = [
        "open_time", "open", "high", "low", "close", "volume", "close_time",
        "quote_volume", "trades", "taker_base", "taker_quote", "ignore"
    ]
    df = pd.DataFrame(data, columns=cols)
    if df.empty:
        return df
    for col in ["open", "high", "low", "close", "volume", "quote_volume", "taker_base", "taker_quote"]:
        df[col] = pd.to_numeric(df[col], errors="coerce")
    df["open_time"] = pd.to_datetime(df["open_time"], unit="ms", utc=True)
    df["close_time"] = pd.to_datetime(df["close_time"], unit="ms", utc=True)
    return df


@st.cache_data(ttl=10, show_spinner=False)
def fetch_mark_price(symbol: str) -> Tuple[Optional[float], str]:
    symbol = normalize_symbol(symbol)
    data = http_get_json(f"{BINANCE_FAPI}/fapi/v1/premiumIndex", params={"symbol": symbol}, timeout=7)
    price = safe_float(data.get("markPrice"), default=np.nan)
    if math.isnan(price):
        return None, "binance_futures_mark_price_unavailable"
    return price, "binance_futures_mark_price"

# =============================================================================
# Indicator + signal generation
# =============================================================================


def ema(series: pd.Series, span: int) -> pd.Series:
    return series.ewm(span=span, adjust=False).mean()


def rsi(series: pd.Series, period: int = 14) -> pd.Series:
    delta = series.diff()
    up = delta.clip(lower=0)
    down = -delta.clip(upper=0)
    roll_up = up.ewm(alpha=1 / period, adjust=False).mean()
    roll_down = down.ewm(alpha=1 / period, adjust=False).mean()
    rs = roll_up / roll_down.replace(0, np.nan)
    return 100 - (100 / (1 + rs))


def true_range(df: pd.DataFrame) -> pd.Series:
    prev_close = df["close"].shift(1)
    return pd.concat([
        df["high"] - df["low"],
        (df["high"] - prev_close).abs(),
        (df["low"] - prev_close).abs(),
    ], axis=1).max(axis=1)


def compute_features(df: pd.DataFrame) -> Dict[str, float]:
    if df.empty or len(df) < 35:
        raise ValueError("not enough candles")
    close = df["close"]
    high = df["high"]
    low = df["low"]
    volume = df["volume"]
    price = float(close.iloc[-1])
    ema8 = float(ema(close, 8).iloc[-1])
    ema21 = float(ema(close, 21).iloc[-1])
    ema55 = float(ema(close, 55).iloc[-1]) if len(close) >= 55 else float(ema(close, 34).iloc[-1])
    rsi14 = float(rsi(close, 14).iloc[-1])
    atr14 = float(true_range(df).rolling(14).mean().iloc[-1])
    ret_3 = float((close.iloc[-1] / close.iloc[-4] - 1) * 100) if len(close) > 4 else 0.0
    ret_6 = float((close.iloc[-1] / close.iloc[-7] - 1) * 100) if len(close) > 7 else 0.0
    ret_12 = float((close.iloc[-1] / close.iloc[-13] - 1) * 100) if len(close) > 13 else 0.0
    vol_mean = float(volume.tail(40).mean())
    vol_std = float(volume.tail(40).std(ddof=0)) or 1.0
    vol_z = float((volume.iloc[-1] - vol_mean) / vol_std)
    quote_volume = float(df["quote_volume"].tail(20).mean()) if "quote_volume" in df else 0.0
    candle_range = float((high.iloc[-1] - low.iloc[-1]) / price * 100)
    body = float((close.iloc[-1] - df["open"].iloc[-1]) / price * 100)
    atr_pct = float(atr14 / price * 100) if price else 0.0
    trend_fast = float((ema8 - ema21) / price * 100)
    trend_slow = float((ema21 - ema55) / price * 100)
    return {
        "price": price,
        "ema8": ema8,
        "ema21": ema21,
        "ema55": ema55,
        "rsi14": rsi14,
        "atr_pct": atr_pct,
        "ret_3": ret_3,
        "ret_6": ret_6,
        "ret_12": ret_12,
        "vol_z": vol_z,
        "quote_volume": quote_volume,
        "candle_range": candle_range,
        "body": body,
        "trend_fast": trend_fast,
        "trend_slow": trend_slow,
        "data_age_seconds": max(0.0, (now_utc() - df["close_time"].iloc[-1].to_pydatetime()).total_seconds()),
    }


def raw_signal_from_features(symbol: str, f: Dict[str, float]) -> Dict[str, Any]:
    # Scores are intentionally transparent and deterministic: the learning layer is the adaptive part.
    long_score = 0.0
    short_score = 0.0
    flat_score = 0.0

    # Trend and momentum agreement
    long_score += max(0, f["trend_fast"]) * 18 + max(0, f["trend_slow"]) * 10
    short_score += max(0, -f["trend_fast"]) * 18 + max(0, -f["trend_slow"]) * 10
    long_score += max(0, f["ret_3"]) * 4 + max(0, f["ret_6"]) * 2.6 + max(0, f["ret_12"]) * 1.2
    short_score += max(0, -f["ret_3"]) * 4 + max(0, -f["ret_6"]) * 2.6 + max(0, -f["ret_12"]) * 1.2

    # RSI regime
    if 48 <= f["rsi14"] <= 68:
        long_score += 7
    if 32 <= f["rsi14"] <= 52:
        short_score += 7
    if f["rsi14"] > 74:
        short_score += 5
    if f["rsi14"] < 26:
        long_score += 5

    # Volume confirms movement; low volatility favors flat
    if f["vol_z"] > 0.75:
        if f["body"] > 0:
            long_score += 4
        elif f["body"] < 0:
            short_score += 4
    if abs(f["trend_fast"]) < 0.035 and abs(f["ret_6"]) < 0.12 and f["atr_pct"] < 0.28:
        flat_score += 18
    if f["vol_z"] < -0.8 and f["candle_range"] < max(0.10, f["atr_pct"] * 0.65):
        flat_score += 8

    # Choose direction.
    directional_edge = long_score - short_score
    max_directional = max(long_score, short_score)
    if flat_score >= max_directional * 0.92 and flat_score >= 12:
        direction = "FLAT"
        edge = flat_score - max_directional
    elif directional_edge > 2.5:
        direction = "UP"
        edge = directional_edge
    elif directional_edge < -2.5:
        direction = "DOWN"
        edge = -directional_edge
    else:
        direction = "FLAT"
        edge = flat_score + max(0, 5 - abs(directional_edge))

    # Setup labels: simple and auditable.
    if direction == "UP" and f["trend_fast"] > 0 and f["ret_6"] > 0:
        setup = "A"  # trend continuation up
        setup_name = "A: bullish continuation"
    elif direction == "DOWN" and f["trend_fast"] < 0 and f["ret_6"] < 0:
        setup = "B"  # bearish continuation/breakdown
        setup_name = "B: bearish continuation"
    elif direction in ("UP", "DOWN") and (f["rsi14"] < 30 or f["rsi14"] > 70):
        setup = "C"  # mean reversion / exhaustion
        setup_name = "C: exhaustion reversion"
    else:
        setup = "D"  # low edge / flat / mixed
        setup_name = "D: mixed or flat regime"

    confidence = clamp(52 + edge * 2.4 + min(8, max(0, f["vol_z"]) * 1.8), 50, 88)
    if direction == "FLAT":
        confidence = clamp(55 + min(22, flat_score * 1.1), 52, 78)

    rationale = (
        f"{display_symbol(symbol)} {direction} from {setup_name}. "
        f"EMA trend={f['trend_fast']:+.3f}%, momentum6={f['ret_6']:+.3f}%, "
        f"RSI14={f['rsi14']:.1f}, ATR={f['atr_pct']:.3f}%, volume z={f['vol_z']:+.2f}."
    )
    return {
        "symbol": normalize_symbol(symbol),
        "direction": direction,
        "setup": setup,
        "setup_name": setup_name,
        "raw_confidence": round(confidence, 2),
        "entry_price": float(f["price"]),
        "model_used": DEFAULT_MODEL,
        "rationale": rationale,
        "features": f,
    }

# =============================================================================
# Memory + optimization
# =============================================================================


def closed_verified_df() -> pd.DataFrame:
    df = load_predictions("status='CLOSED' AND verification_status='VERIFIED' AND hit IS NOT NULL")
    if df.empty:
        return df
    for col in ["raw_confidence", "optimized_confidence", "alpha_score", "entry_price", "close_price", "return_pct", "memory_score_at_open", "optimization_adjustment"]:
        if col in df:
            df[col] = pd.to_numeric(df[col], errors="coerce")
    df["hit"] = pd.to_numeric(df["hit"], errors="coerce").fillna(0).astype(int)
    return df


def summarize_group(df: pd.DataFrame, keys: List[str]) -> pd.DataFrame:
    if df.empty:
        return pd.DataFrame()
    g = df.groupby(keys, dropna=False)
    rows = []
    for key, part in g:
        if not isinstance(key, tuple):
            key = (key,)
        part_sorted = part.sort_values("created_at", ascending=False)
        hits = part["hit"].astype(int)
        returns = pd.to_numeric(part["return_pct"], errors="coerce")
        last10 = part.sort_values("created_at", ascending=False).head(10)["hit"].astype(int).tolist()
        streak = 0
        last_sorted = part.sort_values("created_at", ascending=False)["hit"].astype(int).tolist()
        if last_sorted:
            first = last_sorted[0]
            for v in last_sorted:
                if v == first:
                    streak += 1 if v == 1 else -1
                else:
                    break
        row = {k: v for k, v in zip(keys, key)}
        row.update({
            "predictions": int(len(part)),
            "closed": int(len(part)),
            "wins": int(hits.sum()),
            "losses": int((hits == 0).sum()),
            "win_rate": float(hits.mean() * 100) if len(hits) else np.nan,
            "avg_return": float(returns.mean()) if len(returns) else np.nan,
            "median_return": float(returns.median()) if len(returns) else np.nan,
            "best_return": float(returns.max()) if len(returns) else np.nan,
            "worst_return": float(returns.min()) if len(returns) else np.nan,
            "current_streak": int(streak),
            "last_10": "".join("✅" if x else "✗" for x in last10),
            "last_seen": str(part_sorted["created_at"].iloc[0]) if not part_sorted.empty else "",
        })
        row["status"] = memory_status(row["closed"], row["win_rate"], row["avg_return"])
        rows.append(row)
    out = pd.DataFrame(rows)
    if not out.empty:
        out = out.sort_values(["status", "win_rate", "avg_return", "closed"], ascending=[True, False, False, False])
    return out


def memory_status(closed: int, win_rate: float, avg_return: float, min_closed: int = 8) -> str:
    closed = int(closed or 0)
    win_rate = safe_float(win_rate, 0)
    avg_return = safe_float(avg_return, 0)
    if closed >= min_closed and win_rate >= 70 and avg_return > 0:
        return "PROMOTE"
    if closed >= 5 and 55 <= win_rate < 70 and avg_return >= -0.25:
        return "WATCH"
    if closed >= 5 and (win_rate < 45 or avg_return < -1.0):
        return "FILTER"
    if closed >= 8 and win_rate < 35 and avg_return < -1.5:
        return "BLOCK"
    return "NEUTRAL"


def directional_hit_rates(df: pd.DataFrame) -> pd.DataFrame:
    if df.empty:
        return pd.DataFrame()
    rows = []
    for symbol, part in df.groupby("symbol"):
        row = {"symbol": symbol}
        for d in ["UP", "DOWN", "FLAT"]:
            p = part[part["direction"] == d]
            row[f"{d}_closed"] = len(p)
            row[f"{d}_hit_rate"] = float(p["hit"].mean() * 100) if len(p) else np.nan
        rows.append(row)
    return pd.DataFrame(rows)


def get_group_stats(symbol: str, direction: Optional[str] = None, setup: Optional[str] = None, model: Optional[str] = None) -> Dict[str, Any]:
    df = closed_verified_df()
    if df.empty:
        return {"closed": 0, "win_rate": np.nan, "avg_return": np.nan, "status": "NEUTRAL", "current_streak": 0}
    filt = df["symbol"].eq(normalize_symbol(symbol))
    if direction:
        filt &= df["direction"].eq(direction)
    if setup:
        filt &= df["setup"].eq(setup)
    if model:
        filt &= df["model_used"].eq(model)
    part = df[filt]
    if part.empty:
        return {"closed": 0, "win_rate": np.nan, "avg_return": np.nan, "status": "NEUTRAL", "current_streak": 0}
    summary = summarize_group(part, ["symbol"]).iloc[0].to_dict()
    return summary


def has_recent_duplicate(symbol: str, direction: str, setup: str, minutes: int) -> bool:
    cutoff = (now_utc() - timedelta(minutes=minutes)).isoformat(timespec="seconds")
    with conn() as c:
        n = c.execute(
            """
            SELECT COUNT(*) AS n FROM predictions
            WHERE status='OPEN' AND symbol=? AND direction=? AND setup=? AND created_at >= ?
            """,
            (normalize_symbol(symbol), direction, setup, cutoff),
        ).fetchone()["n"]
    return int(n) > 0


@dataclass
class OptimizerConfig:
    min_closed_for_boost: int = 8
    promote_win_rate: float = 70.0
    promote_avg_return: float = 0.0
    boost_strength: float = 1.0
    penalty_strength: float = 1.0
    duplicate_window_minutes: int = 45
    stale_seconds: int = 1800


def grade_from_score(score: float) -> str:
    if score >= 88:
        return "A+"
    if score >= 78:
        return "A"
    if score >= 68:
        return "B"
    if score >= 58:
        return "C"
    if score >= 45:
        return "D"
    return "X"


def action_from_grade_status(grade: str, status: str, optimized_conf: float) -> str:
    if status == "BLOCK" or grade == "X":
        return "BLOCK"
    if status == "FILTER" or optimized_conf < 56:
        return "SKIP"
    if status == "PROMOTE" and grade in {"A+", "A", "B"}:
        return "PROMOTE"
    if grade in {"A+", "A"}:
        return "WATCH"
    return "WATCH" if optimized_conf >= 60 else "SKIP"


def optimize_signal(sig: Dict[str, Any], cfg: OptimizerConfig) -> Dict[str, Any]:
    symbol = sig["symbol"]
    direction = sig["direction"]
    setup = sig["setup"]
    model = sig.get("model_used", DEFAULT_MODEL)
    raw = float(sig["raw_confidence"])
    f = sig.get("features", {}) or {}

    symbol_stats = get_group_stats(symbol)
    dir_stats = get_group_stats(symbol, direction=direction)
    setup_stats = get_group_stats(symbol, direction=direction, setup=setup)
    model_stats = get_group_stats(symbol, direction=direction, setup=setup, model=model)

    notes = []
    boost = 0.0
    penalty = 0.0
    memory_score = 50.0

    # Symbol-level status.
    symbol_closed = int(symbol_stats.get("closed", 0) or 0)
    symbol_wr = safe_float(symbol_stats.get("win_rate"), 50)
    symbol_avg = safe_float(symbol_stats.get("avg_return"), 0)
    symbol_status = memory_status(symbol_closed, symbol_wr, symbol_avg, cfg.min_closed_for_boost)

    if symbol_closed >= cfg.min_closed_for_boost:
        edge = (symbol_wr - 50) * 0.20 + symbol_avg * 2.2
        memory_score += edge
        if symbol_wr >= cfg.promote_win_rate and symbol_avg > cfg.promote_avg_return:
            b = clamp(edge * 0.55, 2.0, 12.0) * cfg.boost_strength
            boost += b
            notes.append(f"symbol boost {b:+.1f}: {display_symbol(symbol)} has {symbol_closed} verified closes, {symbol_wr:.1f}% win rate, {symbol_avg:+.2f}% avg return")
        elif symbol_wr < 45 or symbol_avg < -1.0:
            p = clamp(abs(edge) * 0.6, 2.0, 14.0) * cfg.penalty_strength
            penalty += p
            notes.append(f"symbol penalty {p:.1f}: weak verified symbol memory")
    else:
        p = clamp((cfg.min_closed_for_boost - symbol_closed) * 0.45, 0.5, 4.0) * cfg.penalty_strength
        penalty += p
        notes.append(f"low-sample penalty {p:.1f}: only {symbol_closed} verified symbol closes")

    # Direction/setup pair has more influence than generic symbol memory.
    for label, stats, weight in [
        ("direction", dir_stats, 0.75),
        ("setup", setup_stats, 1.05),
        ("model/setup", model_stats, 0.45),
    ]:
        closed = int(stats.get("closed", 0) or 0)
        wr = safe_float(stats.get("win_rate"), 50)
        avg = safe_float(stats.get("avg_return"), 0)
        if closed >= max(4, cfg.min_closed_for_boost // 2):
            edge = (wr - 50) * 0.18 + avg * 2.0
            memory_score += edge * weight
            if wr >= 68 and avg > 0:
                b = clamp(edge * weight * 0.42, 1.0, 10.0) * cfg.boost_strength
                boost += b
                notes.append(f"{label} boost {b:+.1f}: {closed} verified closes, {wr:.1f}% hit, {avg:+.2f}% avg")
            elif wr < 44 or avg < -0.8:
                p = clamp(abs(edge) * weight * 0.55, 1.0, 10.0) * cfg.penalty_strength
                penalty += p
                notes.append(f"{label} penalty {p:.1f}: weak closed-memory pattern")

    # Current streak.
    streak = int(setup_stats.get("current_streak", 0) or 0)
    if streak >= 3:
        b = min(4.5, streak * 0.8) * cfg.boost_strength
        boost += b
        notes.append(f"streak boost {b:+.1f}: recent verified wins in this pattern")
    elif streak <= -2:
        p = min(6.0, abs(streak) * 1.3) * cfg.penalty_strength
        penalty += p
        notes.append(f"streak penalty {p:.1f}: recent verified losses in this pattern")

    # Duplicate and stale safeguards.
    duplicate = has_recent_duplicate(symbol, direction, setup, cfg.duplicate_window_minutes)
    if duplicate:
        p = 7.0 * cfg.penalty_strength
        penalty += p
        notes.append(f"duplicate penalty {p:.1f}: similar open signal exists within {cfg.duplicate_window_minutes} min")

    stale = safe_float(f.get("data_age_seconds"), 0) > cfg.stale_seconds
    if stale:
        p = 8.0 * cfg.penalty_strength
        penalty += p
        notes.append("stale-data penalty 8.0: candle data is too old")

    adjustment = boost - penalty
    optimized = clamp(raw + adjustment, 35, 95)

    # Alpha score is not just confidence; it also punishes low evidence.
    sample_factor = clamp(symbol_closed / max(1, cfg.min_closed_for_boost), 0.0, 1.25)
    alpha = clamp(
        optimized * 0.72 + clamp(memory_score, 0, 100) * 0.28 + min(8, max(0, symbol_avg)) - (1 - min(sample_factor, 1)) * 7,
        0,
        100,
    )
    grade = grade_from_score(alpha)
    action = action_from_grade_status(grade, symbol_status, optimized)

    return {
        **sig,
        "optimized_confidence": round(optimized, 2),
        "alpha_score": round(alpha, 2),
        "alpha_grade": grade,
        "action_label": action,
        "memory_score_at_open": round(clamp(memory_score, 0, 100), 2),
        "optimization_adjustment": round(adjustment, 2),
        "optimizer_notes": "; ".join(notes) if notes else "No memory adjustment. Insufficient verified history.",
        "duplicate_flag": int(duplicate),
        "stale_flag": int(stale),
        "memory_status": symbol_status,
    }

# =============================================================================
# Optional LLM rationale via Ollama
# =============================================================================


def call_ollama_rationale(sig: Dict[str, Any], ollama_url: str, ollama_model: str) -> Tuple[str, str]:
    if not ollama_url or not ollama_model:
        return sig.get("rationale", ""), sig.get("model_used", DEFAULT_MODEL)
    prompt = f"""
You are a concise market-research assistant. Rewrite the following deterministic signal into a compact, non-hype rationale.
Do not claim profit is guaranteed. Do not tell the user to buy or sell. Mention that the signal is probabilistic.

Signal:
{json.dumps({k: v for k, v in sig.items() if k not in ['features']}, default=str)}
""".strip()
    try:
        url = ollama_url.rstrip("/") + "/api/generate"
        r = requests.post(
            url,
            json={"model": ollama_model, "prompt": prompt, "stream": False},
            timeout=14,
        )
        r.raise_for_status()
        data = r.json()
        text = str(data.get("response", "")).strip()
        if text:
            return text[:1200], f"ollama:{ollama_model}"
    except Exception as e:
        return sig.get("rationale", "") + f" [Ollama unavailable: {e}]", sig.get("model_used", DEFAULT_MODEL)
    return sig.get("rationale", ""), sig.get("model_used", DEFAULT_MODEL)

# =============================================================================
# Prediction lifecycle
# =============================================================================


def build_prediction_row(sig: Dict[str, Any], horizon_minutes: int, price_source: str = "binance_futures_klines") -> Dict[str, Any]:
    created_at = iso_now()
    pid = stable_id(created_at, sig["symbol"], sig["direction"], sig["setup"], sig.get("model_used", DEFAULT_MODEL), sig.get("entry_price"))
    return {
        "id": pid,
        "created_at": created_at,
        "closed_at": None,
        "symbol": sig["symbol"],
        "direction": sig["direction"],
        "setup": sig["setup"],
        "raw_confidence": float(sig["raw_confidence"]),
        "optimized_confidence": float(sig["optimized_confidence"]),
        "alpha_score": float(sig["alpha_score"]),
        "alpha_grade": sig["alpha_grade"],
        "action_label": sig["action_label"],
        "model_used": sig.get("model_used", DEFAULT_MODEL),
        "rationale": sig.get("rationale", ""),
        "optimizer_notes": sig.get("optimizer_notes", ""),
        "entry_price": float(sig.get("entry_price") or 0),
        "close_price": None,
        "status": "OPEN",
        "hit": None,
        "return_pct": None,
        "horizon_minutes": int(horizon_minutes),
        "verification_status": "PENDING",
        "price_source": price_source,
        "memory_score_at_open": float(sig.get("memory_score_at_open", 50)),
        "optimization_adjustment": float(sig.get("optimization_adjustment", 0)),
        "duplicate_flag": int(sig.get("duplicate_flag", 0)),
        "stale_flag": int(sig.get("stale_flag", 0)),
        "raw_payload": json.dumps(sig.get("features", {}), default=str),
    }


def directional_return(direction: str, entry: float, close: float, flat_band_pct: float) -> Tuple[float, bool]:
    if not entry or not close:
        return np.nan, False
    move = (close / entry - 1) * 100
    if direction == "UP":
        ret = move
        hit = ret > 0
    elif direction == "DOWN":
        ret = -move
        hit = ret > 0
    else:
        # Positive if price stayed inside flat band; negative if it escaped the band.
        ret = flat_band_pct - abs(move)
        hit = abs(move) <= flat_band_pct
    return float(ret), bool(hit)


def close_due_predictions(flat_band_pct: float = 0.35, force: bool = False) -> Dict[str, int]:
    df = load_predictions("status='OPEN'")
    stats = {"checked": 0, "closed": 0, "failed": 0, "not_due": 0}
    if df.empty:
        return stats
    for _, row in df.iterrows():
        stats["checked"] += 1
        created = parse_dt(row.get("created_at"))
        horizon = int(row.get("horizon_minutes") or 60)
        if not created:
            stats["failed"] += 1
            continue
        due = now_utc() >= created + timedelta(minutes=horizon)
        if not due and not force:
            stats["not_due"] += 1
            continue
        try:
            close_price, source = fetch_mark_price(row["symbol"])
            if not close_price:
                stats["failed"] += 1
                continue
            ret, hit = directional_return(row["direction"], float(row["entry_price"]), close_price, flat_band_pct)
            update_prediction_close(row["id"], iso_now(), float(close_price), hit, ret, "VERIFIED", source)
            stats["closed"] += 1
        except Exception:
            stats["failed"] += 1
    return stats

# =============================================================================
# Rendering helpers
# =============================================================================


def top_title() -> None:
    st.markdown(f"# {APP_NAME}")
    st.markdown(f"<div class='small-muted'>{APP_SUBTITLE}</div>", unsafe_allow_html=True)
    st.markdown(f"<div class='notice'>{DISCLAIMER}</div>", unsafe_allow_html=True)


def metric_card(label: str, value: str, sub: str = "") -> None:
    st.markdown(
        f"""
        <div class="metric-card">
            <div class="label">{label}</div>
            <div class="value">{value}</div>
            <div class="sub">{sub}</div>
        </div>
        """,
        unsafe_allow_html=True,
    )


def format_history_df(df: pd.DataFrame) -> pd.DataFrame:
    if df.empty:
        return df
    out = df.copy()
    out["contract"] = out["symbol"].apply(display_symbol)
    out["created"] = pd.to_datetime(out["created_at"], errors="coerce").dt.strftime("%m/%d %H:%M")
    out["raw"] = out["raw_confidence"].map(lambda x: pct_plain(x, 0))
    out["opt"] = out["optimized_confidence"].map(lambda x: pct_plain(x, 0))
    out["alpha"] = out["alpha_score"].map(lambda x: f"{safe_float(x):.0f} {''}") + out["alpha_grade"].astype(str)
    out["hit"] = out["hit"].map(lambda x: "—" if pd.isna(x) else ("✅" if int(x) == 1 else "✗"))
    out["return"] = out["return_pct"].map(lambda x: pct(x, 2))
    out["flags"] = out.apply(lambda r: " ".join([x for x in ["DUP" if int(r.get("duplicate_flag") or 0) else "", "STALE" if int(r.get("stale_flag") or 0) else ""] if x]), axis=1)
    cols = ["created", "contract", "direction", "setup", "raw", "opt", "alpha", "action_label", "status", "hit", "return", "model_used", "verification_status", "flags", "rationale", "optimizer_notes"]
    return out[[c for c in cols if c in out.columns]]


def style_memory_status(v: str) -> str:
    return {
        "PROMOTE": "🟢 PROMOTE",
        "WATCH": "🔵 WATCH",
        "NEUTRAL": "⚪ NEUTRAL",
        "FILTER": "🟠 FILTER",
        "BLOCK": "🔴 BLOCK",
    }.get(str(v), str(v))


def format_memory_df(df: pd.DataFrame) -> pd.DataFrame:
    if df.empty:
        return df
    out = df.copy()
    if "symbol" in out:
        out["contract"] = out["symbol"].apply(display_symbol)
    out["win_rate"] = out["win_rate"].map(lambda x: pct_plain(x, 1))
    for col in ["avg_return", "median_return", "best_return", "worst_return"]:
        if col in out:
            out[col] = out[col].map(lambda x: pct(x, 3))
    if "status" in out:
        out["status"] = out["status"].map(style_memory_status)
    preferred = ["contract", "direction", "setup", "model_used", "predictions", "closed", "wins", "losses", "win_rate", "avg_return", "median_return", "best_return", "worst_return", "current_streak", "last_10", "status", "last_seen"]
    return out[[c for c in preferred if c in out.columns]]

# =============================================================================
# Sidebar settings
# =============================================================================


def sidebar_config() -> Dict[str, Any]:
    st.sidebar.markdown("## 🧠 MEMBRA")
    st.sidebar.caption("Prediction memory engine")
    page = st.sidebar.radio(
        "Navigate",
        ["Dashboard", "Scanner", "History", "Memory", "Diagnostics", "Settings"],
        index=0,
    )
    st.sidebar.divider()
    st.sidebar.markdown("### Engine settings")
    horizon = st.sidebar.number_input("Prediction close horizon, minutes", min_value=5, max_value=1440, value=60, step=5)
    flat_band = st.sidebar.number_input("FLAT hit band, %", min_value=0.05, max_value=5.0, value=0.35, step=0.05)
    min_closed = st.sidebar.number_input("Min closes for memory boost", min_value=2, max_value=100, value=8, step=1)
    duplicate_window = st.sidebar.number_input("Duplicate detection window, minutes", min_value=5, max_value=360, value=45, step=5)
    boost_strength = st.sidebar.slider("Boost strength", 0.0, 2.5, 1.0, 0.1)
    penalty_strength = st.sidebar.slider("Penalty strength", 0.0, 2.5, 1.0, 0.1)
    st.sidebar.divider()
    st.sidebar.markdown("### Optional Ollama rationale")
    use_ollama = st.sidebar.toggle("Use Ollama if reachable", value=False)
    ollama_url = st.sidebar.text_input("Ollama URL", value=os.getenv("OLLAMA_URL", "http://localhost:11434"))
    ollama_model = st.sidebar.text_input("Ollama model", value=os.getenv("OLLAMA_MODEL", "llama3.1"))
    return {
        "page": page,
        "horizon": int(horizon),
        "flat_band": float(flat_band),
        "optimizer": OptimizerConfig(
            min_closed_for_boost=int(min_closed),
            boost_strength=float(boost_strength),
            penalty_strength=float(penalty_strength),
            duplicate_window_minutes=int(duplicate_window),
        ),
        "use_ollama": bool(use_ollama),
        "ollama_url": ollama_url,
        "ollama_model": ollama_model,
    }

# =============================================================================
# Pages
# =============================================================================


def page_dashboard(cfg: Dict[str, Any]) -> None:
    top_title()
    close_stats = close_due_predictions(flat_band_pct=cfg["flat_band"], force=False)
    if close_stats["closed"]:
        st.success(f"Closed and verified {close_stats['closed']} due prediction(s).")

    counts = count_predictions()
    df_closed = closed_verified_df()
    total_closed = len(df_closed)
    overall_wr = safe_float(df_closed["hit"].mean() * 100, 0) if total_closed else np.nan
    avg_ret = safe_float(df_closed["return_pct"].mean(), np.nan) if total_closed else np.nan

    mem_symbol = summarize_group(df_closed, ["symbol"]) if not df_closed.empty else pd.DataFrame()
    best_symbol = "—"
    worst_symbol = "—"
    best_setup = "—"
    worst_setup = "—"
    if not mem_symbol.empty:
        mem_rank = mem_symbol.sort_values(["avg_return", "win_rate", "closed"], ascending=[False, False, False])
        best_symbol = f"{display_symbol(mem_rank.iloc[0]['symbol'])} {pct(mem_rank.iloc[0]['avg_return'], 2)}"
        worst = mem_symbol.sort_values(["avg_return", "win_rate"], ascending=[True, True]).iloc[0]
        worst_symbol = f"{display_symbol(worst['symbol'])} {pct(worst['avg_return'], 2)}"
    mem_setup = summarize_group(df_closed, ["setup"]) if not df_closed.empty else pd.DataFrame()
    if not mem_setup.empty:
        s1 = mem_setup.sort_values(["avg_return", "win_rate"], ascending=[False, False]).iloc[0]
        s2 = mem_setup.sort_values(["avg_return", "win_rate"], ascending=[True, True]).iloc[0]
        best_setup = f"Setup {s1['setup']} {pct(s1['avg_return'], 2)}"
        worst_setup = f"Setup {s2['setup']} {pct(s2['avg_return'], 2)}"

    c1, c2, c3, c4 = st.columns(4)
    with c1:
        metric_card("Total predictions", str(counts["TOTAL"]), f"Open {counts['OPEN']} / Closed {counts['CLOSED']}")
    with c2:
        metric_card("Verified win rate", pct_plain(overall_wr, 1), f"Closed verified {total_closed}")
    with c3:
        metric_card("Average return", pct(avg_ret, 3), "Directional return after close")
    with c4:
        metric_card("Optimizer health", "PASS" if total_closed >= 1 else "WAIT", "Learning uses closed verified outcomes only")

    c5, c6, c7, c8 = st.columns(4)
    with c5:
        metric_card("Best symbol", best_symbol, "By average verified return")
    with c6:
        metric_card("Worst symbol", worst_symbol, "Candidate for filter review")
    with c7:
        metric_card("Best setup", best_setup, "Verified closed history")
    with c8:
        metric_card("Weakest setup", worst_setup, "Verified closed history")

    st.markdown("## Top memory symbols")
    if mem_symbol.empty:
        st.info("No closed verified memory yet. Generate scanner predictions, save them, then close/verify after the horizon.")
    else:
        view = mem_symbol.sort_values(["status", "win_rate", "avg_return"], ascending=[True, False, False]).head(12)
        st.dataframe(format_memory_df(view), use_container_width=True, hide_index=True)

    st.markdown("## Last 10 closed predictions")
    hist = load_predictions("status='CLOSED'").head(10)
    st.dataframe(format_history_df(hist), use_container_width=True, hide_index=True)


def get_scan_universe(custom_symbols: str, top_n: int) -> List[str]:
    if custom_symbols.strip():
        raw = [normalize_symbol(x) for x in custom_symbols.split(",") if x.strip()]
        return list(dict.fromkeys(raw))[:max(1, top_n)]
    try:
        valid = set(fetch_exchange_symbols())
        tickers = fetch_24h_tickers()
        tickers = tickers[tickers["symbol"].isin(valid)]
        tickers = tickers.sort_values("quoteVolume", ascending=False).head(top_n)
        return tickers["symbol"].tolist()
    except Exception:
        return [normalize_symbol(x) for x in DEFAULT_SYMBOLS.split(",")][:top_n]


def scan_symbols(symbols: List[str], interval: str, limit: int, cfg: Dict[str, Any]) -> Tuple[pd.DataFrame, List[str]]:
    rows = []
    errors = []
    valid_symbols = set()
    try:
        valid_symbols = set(fetch_exchange_symbols())
    except Exception:
        valid_symbols = set()

    progress = st.progress(0, text="Scanning market data…")
    for i, sym in enumerate(symbols):
        sym = normalize_symbol(sym)
        progress.progress((i + 1) / max(1, len(symbols)), text=f"Scanning {sym}…")
        if valid_symbols and sym not in valid_symbols:
            errors.append(f"{sym}: not listed as Binance USDT perpetual or not trading")
            continue
        try:
            k = fetch_klines(sym, interval=interval, limit=limit)
            f = compute_features(k)
            sig = raw_signal_from_features(sym, f)
            opt = optimize_signal(sig, cfg["optimizer"])
            if cfg["use_ollama"]:
                rationale, model_used = call_ollama_rationale(opt, cfg["ollama_url"], cfg["ollama_model"])
                opt["rationale"] = rationale
                opt["model_used"] = model_used
            rows.append(opt)
        except Exception as e:
            errors.append(f"{sym}: {e}")
    progress.empty()
    if not rows:
        return pd.DataFrame(), errors
    df = pd.DataFrame(rows)
    df = df.sort_values(["action_label", "alpha_score", "optimized_confidence"], ascending=[True, False, False])
    return df, errors


def page_scanner(cfg: Dict[str, Any]) -> None:
    top_title()
    st.markdown("## Scanner")
    st.caption("Generates deterministic market predictions, then applies closed-memory optimization. Saving a candidate creates an OPEN prediction record.")
    colA, colB, colC, colD = st.columns([2.2, 1, 1, 1])
    with colA:
        custom = st.text_input("Contracts to scan, comma-separated", value=DEFAULT_SYMBOLS)
    with colB:
        interval = st.selectbox("Candle interval", ["1m", "3m", "5m", "15m", "30m", "1h"], index=2)
    with colC:
        limit = st.number_input("Candles", min_value=60, max_value=500, value=180, step=20)
    with colD:
        top_n = st.number_input("Max symbols", min_value=1, max_value=80, value=20, step=1)

    scan_button = st.button("Run scanner", type="primary", use_container_width=True)
    if scan_button:
        symbols = get_scan_universe(custom, int(top_n))
        df, errors = scan_symbols(symbols, interval, int(limit), cfg)
        st.session_state["scan_df"] = df
        st.session_state["scan_errors"] = errors

    df = st.session_state.get("scan_df", pd.DataFrame())
    errors = st.session_state.get("scan_errors", [])

    if errors:
        with st.expander(f"Skipped / failed symbols ({len(errors)})"):
            st.write("\n".join(errors))

    if df is None or df.empty:
        st.info("Run the scanner to generate signal candidates. No fabricated fallback data is used.")
        return

    display = df.copy()
    display["contract"] = display["symbol"].apply(display_symbol)
    display["raw"] = display["raw_confidence"].map(lambda x: pct_plain(x, 0))
    display["optimized"] = display["optimized_confidence"].map(lambda x: pct_plain(x, 0))
    display["alpha"] = display.apply(lambda r: f"{r['alpha_score']:.0f} / {r['alpha_grade']}", axis=1)
    display["memory"] = display["memory_status"].astype(str)
    display["entry"] = display["entry_price"].map(lambda x: f"{safe_float(x):,.8g}")
    cols = ["contract", "direction", "setup", "raw", "optimized", "alpha", "action_label", "memory", "entry", "rationale", "optimizer_notes"]
    st.dataframe(display[cols], use_container_width=True, hide_index=True)

    st.markdown("### Save candidates to prediction history")
    selectable = display["contract"].tolist()
    chosen = st.multiselect("Select candidates to save as OPEN predictions", selectable, default=selectable[: min(3, len(selectable))])
    if st.button("Save selected OPEN predictions", use_container_width=True):
        saved = 0
        for _, r in df.iterrows():
            if display_symbol(r["symbol"]) in chosen:
                row = build_prediction_row(r.to_dict(), cfg["horizon"], price_source="binance_futures_klines")
                insert_prediction(row)
                saved += 1
        st.success(f"Saved {saved} open prediction(s). They will be eligible for verification after {cfg['horizon']} minutes.")


def page_history(cfg: Dict[str, Any]) -> None:
    top_title()
    st.markdown("## Prediction History")
    c1, c2, c3, c4 = st.columns(4)
    with c1:
        status = st.selectbox("Status", ["ALL", "OPEN", "CLOSED"], index=0)
    with c2:
        direction = st.selectbox("Direction", ["ALL", "UP", "DOWN", "FLAT"], index=0)
    with c3:
        setup = st.selectbox("Setup", ["ALL", "A", "B", "C", "D"], index=0)
    with c4:
        symbol = st.text_input("Symbol contains", value="")

    where = []
    params: List[Any] = []
    if status != "ALL":
        where.append("status=?")
        params.append(status)
    if direction != "ALL":
        where.append("direction=?")
        params.append(direction)
    if setup != "ALL":
        where.append("setup=?")
        params.append(setup)
    if symbol.strip():
        where.append("symbol LIKE ?")
        params.append(f"%{normalize_symbol(symbol)}%")
    df = load_predictions(" AND ".join(where), tuple(params))

    b1, b2 = st.columns(2)
    with b1:
        if st.button("Close due predictions now", use_container_width=True):
            stats = close_due_predictions(flat_band_pct=cfg["flat_band"], force=False)
            st.success(f"Checked {stats['checked']}; closed {stats['closed']}; not due {stats['not_due']}; failed {stats['failed']}.")
            st.rerun()
    with b2:
        if st.button("Force-close all OPEN predictions", use_container_width=True):
            stats = close_due_predictions(flat_band_pct=cfg["flat_band"], force=True)
            st.warning(f"Force close complete. Checked {stats['checked']}; closed {stats['closed']}; failed {stats['failed']}.")
            st.rerun()

    if df.empty:
        st.info("No matching prediction history yet.")
    else:
        st.dataframe(format_history_df(df), use_container_width=True, hide_index=True)
        csv_bytes = df.to_csv(index=False).encode("utf-8")
        st.download_button("Download history CSV", data=csv_bytes, file_name="membra_prediction_history.csv", mime="text/csv")


def page_memory(cfg: Dict[str, Any]) -> None:
    top_title()
    st.markdown("## Memory")
    df = closed_verified_df()
    if df.empty:
        st.info("No closed verified predictions yet. Memory will stay empty until OPEN predictions are closed and verified.")
        return

    group_mode = st.selectbox(
        "Memory grouping",
        ["Symbol", "Symbol + Direction", "Symbol + Direction + Setup", "Model", "Model + Setup"],
        index=0,
    )
    keys_map = {
        "Symbol": ["symbol"],
        "Symbol + Direction": ["symbol", "direction"],
        "Symbol + Direction + Setup": ["symbol", "direction", "setup"],
        "Model": ["model_used"],
        "Model + Setup": ["model_used", "setup"],
    }
    mem = summarize_group(df, keys_map[group_mode])
    if mem.empty:
        st.info("No memory rows for that grouping.")
    else:
        st.dataframe(format_memory_df(mem), use_container_width=True, hide_index=True)

    st.markdown("### Directional hit rates by symbol")
    dhr = directional_hit_rates(df)
    if not dhr.empty:
        view = dhr.copy()
        view["contract"] = view["symbol"].apply(display_symbol)
        for col in ["UP_hit_rate", "DOWN_hit_rate", "FLAT_hit_rate"]:
            view[col] = view[col].map(lambda x: pct_plain(x, 1))
        cols = ["contract", "UP_closed", "UP_hit_rate", "DOWN_closed", "DOWN_hit_rate", "FLAT_closed", "FLAT_hit_rate"]
        st.dataframe(view[cols], use_container_width=True, hide_index=True)


def diag_card(status: str, title: str, body: str) -> None:
    color = {"PASS": "green", "WARNING": "warn", "FAIL": "red", "WAIT": "gray"}.get(status, "gray")
    st.markdown(
        f"""
        <div class="metric-card">
          <div class="label"><span class="badge {color}">{status}</span></div>
          <div class="value" style="font-size:1.05rem">{title}</div>
          <div class="sub">{body}</div>
        </div>
        """,
        unsafe_allow_html=True,
    )


def page_diagnostics(cfg: Dict[str, Any]) -> None:
    top_title()
    st.markdown("## Diagnostics")
    counts = count_predictions()
    df_all = load_predictions()
    df_closed = closed_verified_df()
    open_n = counts["OPEN"]
    closed_n = len(df_closed)
    dup_n = int(df_all["duplicate_flag"].fillna(0).sum()) if not df_all.empty else 0
    stale_n = int(df_all["stale_flag"].fillna(0).sum()) if not df_all.empty else 0
    unverified_closed = 0
    if not df_all.empty:
        unverified_closed = len(df_all[(df_all["status"] == "CLOSED") & (df_all["verification_status"] != "VERIFIED")])

    c1, c2, c3 = st.columns(3)
    with c1:
        diag_card("PASS" if counts["TOTAL"] >= 0 else "FAIL", "Prediction logging", f"{counts['TOTAL']} total rows in local SQLite memory.")
    with c2:
        diag_card("PASS" if open_n >= 0 and counts["CLOSED"] >= 0 else "FAIL", "Open/closed separation", f"Open {open_n}, closed {counts['CLOSED']}.")
    with c3:
        diag_card("PASS" if unverified_closed == 0 else "WARNING", "Verification integrity", f"{closed_n} verified closed; {unverified_closed} closed rows not verified.")

    c4, c5, c6 = st.columns(3)
    with c4:
        diag_card("PASS" if closed_n > 0 else "WAIT", "Memory update", "Memory is calculated only from CLOSED + VERIFIED outcomes." if closed_n else "Waiting for closed verified outcomes.")
    with c5:
        diag_card("WARNING" if dup_n else "PASS", "Duplicate detection", f"{dup_n} prediction(s) flagged duplicate.")
    with c6:
        diag_card("WARNING" if stale_n else "PASS", "Stale data detection", f"{stale_n} prediction(s) flagged stale.")

    st.markdown("### Promoted and filtered symbols")
    mem = summarize_group(df_closed, ["symbol"]) if not df_closed.empty else pd.DataFrame()
    if mem.empty:
        st.info("No symbol memory yet.")
    else:
        promoted = mem[mem["status"].eq("PROMOTE")].copy()
        filtered = mem[mem["status"].isin(["FILTER", "BLOCK"])].copy()
        a, b = st.columns(2)
        with a:
            st.markdown("#### Promoted")
            st.dataframe(format_memory_df(promoted), use_container_width=True, hide_index=True)
        with b:
            st.markdown("#### Filtered / blocked")
            st.dataframe(format_memory_df(filtered), use_container_width=True, hide_index=True)

    st.markdown("### Data-source check")
    try:
        syms = fetch_exchange_symbols()
        st.success(f"Binance USDT perpetual symbol list reachable: {len(syms)} active symbols.")
    except Exception as e:
        st.error(f"Market data source unavailable: {e}")


def page_settings(cfg: Dict[str, Any]) -> None:
    top_title()
    st.markdown("## Settings and data management")
    st.write("The main runtime settings are in the sidebar. This section handles memory export/import and reset.")

    df = load_predictions()
    if df.empty:
        st.info("No data in local memory yet.")
    else:
        st.download_button(
            "Export all predictions CSV",
            data=df.to_csv(index=False).encode("utf-8"),
            file_name="membra_alpha_all_predictions.csv",
            mime="text/csv",
            use_container_width=True,
        )

    st.markdown("### Import history CSV")
    uploaded = st.file_uploader("Upload a previously exported MEMBRA prediction CSV", type=["csv"])
    if uploaded is not None:
        try:
            imp = pd.read_csv(uploaded)
            required = {"id", "created_at", "symbol", "direction", "setup", "status"}
            if not required.issubset(set(imp.columns)):
                st.error(f"CSV missing required columns: {sorted(required - set(imp.columns))}")
            elif st.button("Import CSV rows", type="primary"):
                n = 0
                for _, row in imp.iterrows():
                    record = row.to_dict()
                    # Fill any missing columns with safe defaults.
                    base = {
                        "closed_at": None,
                        "raw_confidence": 0,
                        "optimized_confidence": 0,
                        "alpha_score": 0,
                        "alpha_grade": "D",
                        "action_label": "WATCH",
                        "model_used": DEFAULT_MODEL,
                        "rationale": "Imported row",
                        "optimizer_notes": "Imported row",
                        "entry_price": None,
                        "close_price": None,
                        "hit": None,
                        "return_pct": None,
                        "horizon_minutes": cfg["horizon"],
                        "verification_status": "VERIFIED" if str(record.get("status", "")).upper() == "CLOSED" else "PENDING",
                        "price_source": "imported_csv",
                        "memory_score_at_open": 50,
                        "optimization_adjustment": 0,
                        "duplicate_flag": 0,
                        "stale_flag": 0,
                        "raw_payload": "{}",
                    }
                    base.update(record)
                    base["symbol"] = normalize_symbol(str(base["symbol"]))
                    insert_prediction(base)
                    n += 1
                st.success(f"Imported {n} row(s).")
                st.rerun()
        except Exception as e:
            st.error(f"Import failed: {e}")

    st.markdown("### Reset")
    st.warning("Reset permanently clears the local SQLite prediction memory for this app instance.")
    confirm = st.text_input("Type DELETE to enable reset", value="")
    if st.button("Delete all local prediction memory", disabled=(confirm != "DELETE"), use_container_width=True):
        delete_all_data()
        st.success("Local prediction memory cleared.")
        st.rerun()

# =============================================================================
# Main
# =============================================================================


def main() -> None:
    cfg = sidebar_config()
    page = cfg["page"]
    if page == "Dashboard":
        page_dashboard(cfg)
    elif page == "Scanner":
        page_scanner(cfg)
    elif page == "History":
        page_history(cfg)
    elif page == "Memory":
        page_memory(cfg)
    elif page == "Diagnostics":
        page_diagnostics(cfg)
    elif page == "Settings":
        page_settings(cfg)
    else:
        page_dashboard(cfg)

    st.sidebar.divider()
    st.sidebar.caption(DISCLAIMER)


if __name__ == "__main__":
    main()