| 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) |
| |
| |
| 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) |
|
|
