Warrior_EA/research/kit.py

import numpy as np, datetime as dt
R=r"C:/Users/admin/AppData/Roaming/MetaQuotes/Terminal/Common/Files/Warrior_EA/Research/"

def load_rates(sym,tf):
    d=np.genfromtxt(R+f"{sym}_{tf}_rates.csv",delimiter=',',names=True)
    return d['time'],d['open'],d['high'],d['low'],d['close'],d['tickvol'],d['spread']

def atr(h,l,c,n=14):
    pc=np.roll(c,1); pc[0]=c[0]
    tr=np.maximum(h-l,np.maximum(np.abs(h-pc),np.abs(l-pc)))
    out=np.convolve(tr,np.ones(n)/n,mode='full')[:len(tr)]
    out[:n]=tr[:n].mean() if n<len(tr) else tr.mean()
    return out

def sma(x,n):
    out=np.convolve(x,np.ones(n)/n,mode='full')[:len(x)]
    out[:n]=x[:n].mean()
    return out

def features(t,o,h,l,c,v):
    """Scale-free, causal features - everything divided by ATR or expressed as a ratio, so the
    same model is comparable across instruments and across 18 years of price levels."""
    a=atr(h,l,c,14); a=np.where(a>0,a,np.nan)
    F=[]; names=[]
    for k in (1,2,3,5,10,20,50):
        r=(c-np.roll(c,k))/a; r[:k]=0; F.append(r); names.append(f"ret{k}")
    for k in (10,20,50):
        hh=np.array([h[max(0,i-k+1):i+1].max() for i in range(len(h))])
        ll=np.array([l[max(0,i-k+1):i+1].min() for i in range(len(l))])
        F.append(np.where(hh>ll,(c-ll)/(hh-ll),0.5)); names.append(f"donch{k}")
    for k in (10,20,50):
        F.append((c-sma(c,k))/a); names.append(f"sma{k}dist")
    F.append(a/sma(a,50)); names.append("atrratio")
    rng=np.where(h>l,h-l,np.nan)
    F.append((c-o)/rng); names.append("body")
    F.append((h-np.maximum(o,c))/rng); names.append("upwick")
    F.append((np.minimum(o,c)-l)/rng); names.append("dnwick")
    F.append(rng/a); names.append("rangeatr")
    vm=sma(v,50); F.append(np.where(vm>0,v/vm,1.0)); names.append("volratio")
    d=np.roll(c,1)-np.roll(c,2)
    up=np.where(d>0,d,0); dn=np.where(d<0,-d,0)
    rs=sma(up,14)/np.where(sma(dn,14)>0,sma(dn,14),np.nan)
    F.append(100-100/(1+rs)); names.append("rsi14")
    hh=np.array([dt.datetime.fromtimestamp(x,dt.UTC).hour for x in t],dtype=float)
    dw=np.array([dt.datetime.fromtimestamp(x,dt.UTC).weekday() for x in t],dtype=float)
    F.append(np.sin(2*np.pi*hh/24)); names.append("hsin")
    F.append(np.cos(2*np.pi*hh/24)); names.append("hcos")
    F.append(dw); names.append("dow")
    X=np.column_stack(F)
    return np.nan_to_num(X,nan=0.0,posinf=0.0,neginf=0.0),names,a

def barrier_vec(h,l,c,a,sl,tp,H,spread):
    """Vectorised triple-barrier. Stop is tested before target WITHIN a bar, so a bar spanning
    both scores as the loss - implemented as strict tp_idx < sl_idx."""
    n=len(c); INF=np.iinfo(np.int32).max
    lab=np.full(n,2,dtype=np.int8); valid=np.zeros(n,bool)
    risk=sl*a; rew=tp*a
    lTp=c+spread+rew; lSl=c+spread-risk; sTp=c-rew-spread; sSl=c+risk-spread
    CH=200000//max(H,1)+1
    for s in range(0,n,CH):
        e=min(s+CH,n); m=e-s
        if e+H>n: e2=n-H
        else: e2=e
        if e2<=s: break
        mm=e2-s
        wi=np.arange(1,H+1)[None,:]+np.arange(s,e2)[:,None]
        wh=h[wi]; wl=l[wi]
        def first(mask):
            any_=mask.any(axis=1); return np.where(any_,mask.argmax(axis=1),INF)
        lsl=first(wl<=lSl[s:e2,None]); ltp=first(wh>=lTp[s:e2,None])
        ssl=first(wh>=sSl[s:e2,None]); stp=first(wl<=sTp[s:e2,None])
        lw=ltp<lsl; sw=stp<ssl
        seg=np.full(mm,2,dtype=np.int8)
        seg[lw&~sw]=0; seg[sw&~lw]=1
        lab[s:e2]=seg; valid[s:e2]=np.isfinite(a[s:e2])&(a[s:e2]>0)
    return lab,valid