import os, time
from datetime import datetime, timezone
from pathlib import Path
import requests, pandas as pd, numpy as np
import joblib
from sklearn.ensemble import HistGradientBoostingClassifier

ROOT = Path(__file__).parent
DATA = ROOT / "BTCUSD_1m.parquet"
LIVE_DATA = ROOT / "live_candles.parquet"
DELTA = os.getenv("DELTA_URL", "https://api.india.delta.exchange")
SYMBOL = os.getenv("SYMBOL", "BTCUSD")
RECEIVER_URL = os.getenv("RECEIVER_URL", "").rstrip("/")
MOVE, OFFSET, HORIZONS = .0008, .0004, (5, 15, 30)
MODELS = None

def epoch_seconds(value):
    value = float(value)
    if value > 1e14: return int(value / 1e6)
    if value > 1e11: return int(value / 1e3)
    # Some previously cached Delta responses were stored as epoch milliseconds
    # with three digits dropped (e.g. 1786405). Restore them to epoch seconds.
    if 1e6 < value < 1e9: return int(value * 1000)
    return int(value)

def get_candles():
    """Incrementally backfill Delta candles from the newest stored timestamp."""
    now = int(time.time())
    old = pd.read_parquet(LIVE_DATA) if LIVE_DATA.exists() else pd.DataFrame()
    if len(old) and epoch_seconds(old.epoch.max()) < 1577836800:
        old = pd.DataFrame()
    if len(old):
        start = epoch_seconds(old.epoch.max()) + 60
    else:
        base = pd.read_parquet(DATA, columns=["timestamp"])
        start = int(pd.to_datetime(base.timestamp, utc=True).astype("int64").max() // 10**9) + 60
    rows=[]
    while start < now:
        end=min(now, start + 2000*60)
        r=requests.get(DELTA + "/v2/history/candles", params={"resolution":"1m",
            "symbol":SYMBOL,"start":start,"end":end}, timeout=20); r.raise_for_status()
        batch=r.json().get("result", [])
        if not batch: break
        rows.extend(batch)
        latest=max(epoch_seconds(q["time"]) for q in batch)
        if latest < start: break
        start=max(end, latest+60)
    fresh=pd.DataFrame(rows).rename(columns={"time":"epoch"})
    if len(fresh): fresh["epoch"] = fresh["epoch"].map(epoch_seconds)
    live=pd.concat([old,fresh],ignore_index=True) if len(old) or len(fresh) else pd.DataFrame()
    if len(live):
        live=live.drop_duplicates("epoch",keep="last").sort_values("epoch").reset_index(drop=True)
        live.to_parquet(LIVE_DATA,index=False,compression="zstd")
    return live

def make_features(d):
    o,h,l,c,v=[d[k].to_numpy(float) for k in ("open","high","low","close","volume")]
    r=np.maximum(h-l,1e-12); vm=pd.Series(v).rolling(30).median().to_numpy()
    vwap=(pd.Series(c*v).rolling(30).sum()/pd.Series(v).rolling(30).sum()).to_numpy()
    rm=lambda a,w:pd.Series(a).rolling(w).mean().to_numpy()
    ret=c/np.roll(c,1)-1
    ts = d.epoch if "epoch" in d.columns else pd.to_datetime(d.timestamp,utc=True).astype("int64") // 10**9
    x=np.column_stack([ret,c/np.roll(c,3)-1,c/np.roll(c,10)-1,c/np.roll(c,30)-1,
      rm(r,5)/c,rm(r,20)/c,(c-vwap)/c,v/(vm+1e-12),(c-o)/r,
      (h-np.maximum(o,c))/r,(np.minimum(o,c)-l)/r,r/c,rm(ret,5),rm(ret,20),
      pd.to_datetime(ts,unit="s",utc=True).dt.hour.to_numpy()/24])
    return x

def train():
    d=pd.read_parquet(DATA); d["epoch"]=(pd.to_datetime(d.timestamp,utc=True).astype("int64")//10**9); x=make_features(d); c=d.close.to_numpy(float); h=d.high.to_numpy(float); l=d.low.to_numpy(float)
    n=len(c); labels={}; fills={}
    for si,side in enumerate((1,-1)):
        entry=c*(1-OFFSET if side==1 else 1+OFFSET)
        fills[si]=np.r_[((l[1:]<=entry[:-1]) if side==1 else (h[1:]>=entry[:-1])),False]
        for horizon in HORIZONS:
            y=np.zeros(n,bool)
            for i in range(n-horizon-1):
                if fills[si][i]: y[i]=(np.max(h[i+1:i+horizon+1])>=entry[i]*(1+MOVE) if side==1 else np.min(l[i+1:i+horizon+1])<=entry[i]*(1-MOVE))
            labels[si,horizon]=y
    good=np.isfinite(x).all(1); train_ix=np.arange(120,n,5); train_ix=train_ix[good[train_ix]]; models={}
    for si in range(2):
        for horizon in HORIZONS:
            m=HistGradientBoostingClassifier(max_iter=120,learning_rate=.07,max_leaf_nodes=15,min_samples_leaf=80,l2_regularization=2,random_state=41+horizon+si)
            m.fit(x[train_ix],labels[si,horizon][train_ix].astype(int)); models[si,horizon]=m
    return models

def tick():
    global MODELS
    if MODELS is None:
        if (ROOT / "models.joblib").exists():
            try:
                MODELS=joblib.load(ROOT / "models.joblib")
            except Exception:
                MODELS=train()
        else:
            MODELS=train()
    live=get_candles(); hist=pd.read_parquet(DATA).tail(5000)
    hist["epoch"]=(pd.to_datetime(hist.timestamp,utc=True).astype("int64")//10**9)
    d=pd.concat([hist,live],ignore_index=True).drop_duplicates("epoch",keep="last").sort_values("epoch").reset_index(drop=True)
    i=len(d)-2; x=make_features(d); probs=np.zeros((2,3))
    for si in range(2):
        for hi,horizon in enumerate(HORIZONS): probs[si,hi]=MODELS[si,horizon].predict_proba(x[i:i+1])[0,1]
    score=3/(1/(probs+1e-6)).sum(axis=1); si=int(np.argmax(score)); side="LONG" if si==0 else "SHORT"
    row=d.iloc[i]; candle_epoch=epoch_seconds(row.epoch); event={"event":"tick","symbol":SYMBOL,"candle":{"time":datetime.fromtimestamp(candle_epoch,timezone.utc).isoformat(),"epoch":candle_epoch,"open":float(row.open),"high":float(row.high),"low":float(row.low),"close":float(row.close)},"signal":None,"probs":probs.tolist(),"score":float(score[si]),"source":"exact-hybrid-multihorizon"}
    if np.isfinite(x[i]).all(): event["signal"]={"side":side,"price":float(row.close),"score":float(score[si]),"offset":OFFSET,"target":MOVE}
    event["data_last_updated"] = event["candle"]["time"]
    event["live_cache_rows"] = int(len(live))
    if RECEIVER_URL:
        r=requests.post(RECEIVER_URL,json=event,timeout=15); r.raise_for_status()
    return {"ok":True,"sent":bool(RECEIVER_URL),"event":event}

if __name__=="__main__":
    import gradio as gr
    with gr.Blocks() as demo:
        gr.Markdown("# Exact Hybrid BTC Paper Trader")
        out=gr.JSON(); gr.Button("Run exact tick").click(tick,outputs=out,api_name="tick")
    demo.queue().launch(server_name="0.0.0.0",server_port=7860,ssr_mode=False)