generate.py

#!/usr/bin/env python3
"""Generates the six synthetic datasets in data/. Fixed seed: the output is byte for byte
the files in this repository (CI checks it against SHA256SUMS).

    python3 generate.py            # writes into data/
    python3 generate.py --out DIR

Because the data is generated, the true answer is known for every file. That is the point:
you can check that each test result in reports/ matches the truth written below.
"""
import random, math, csv, os
from datetime import datetime, timedelta
import sys
R = random.Random(2026)
EDGE=0.00004  # the published files use this edge for 01

def walk(n, vol, start=100.0, rng=R, common=None, beta=0.0):
    p=[start]; 
    for i in range(1,n):
        c = common[i] if common else 0.0
        p.append(p[-1]*(1+beta*c+rng.gauss(0,vol)))
    return p

# ---------- 01 model scores, 8 assets, 15 min, 4h outcome, weak edge ----------
def f01():
    assets=["BTC","ETH","SOL","BNB","XRP","ADA","DOGE","AVAX"]
    n=4*24*60; t0=datetime(2026,5,1)
    mkt=[R.gauss(0,0.0004) for _ in range(n)]
    rows=[]
    for a in assets:
        z=0.0; sig=[]; px=[R.uniform(1,500)]
        for i in range(n):
            z=0.94*z+R.gauss(0,0.35); sig.append(z)
        for i in range(1,n):
            px.append(px[-1]*(1+mkt[i]+R.gauss(0,0.0003)+EDGE*sig[i-1]))
        for i in range(n-16):
            fwd=(px[i+16]/px[i]-1)*100
            score=50+8*sig[i]+R.gauss(0,6)
            rows.append([(t0+timedelta(minutes=15*i)).strftime("%Y-%m-%d %H:%M"),a,round(score,1),round(px[i],4),round(fwd,4)])
    rows.sort(key=lambda r:(r[0],r[1]))
    w=csv.writer(open("01_model_scores_multi_asset.csv","w",newline=""))
    w.writerow(["timestamp","asset","prediction","price","outcome"]); w.writerows(rows)

# ---------- 02 directional calls, tight TP / wide SL, high 'win rate', no edge ----------
def f02():
    coins={"BTC":60000,"ETH":2600,"SOL":140,"XRP":0.55,"DOGE":0.12,"LINK":13,"AVAX":25,"ADA":0.38,"DOT":4.5,"NEAR":4.8}
    minutes=150*24*60; t0=datetime(2026,3,1)
    paths={c:walk(minutes//5, 0.0020, start=s) for c,s in coins.items()}  # 5-min path
    rows=[]; wins=0
    for k in range(420):
        c=R.choice(list(coins)); i=R.randrange(0,len(paths[c])-300)
        side=R.choice([1,1,1,-1])  # mostly longs, like most channels
        e=paths[c][i]; tp=0.012; sl=0.045
        hit=None
        for j in range(i+1,i+288*3):
            r=(paths[c][j]/e-1)*side
            if r>=tp: hit=1;break
            if r<=-sl: hit=0;break
        fwd=(paths[c][i+288]/e-1)*100  # 24h market move
        ts=t0+timedelta(minutes=5*i)
        wins+= (hit==1)
        rows.append([ts.strftime("%Y-%m-%d %H:%M"),c,"LONG" if side==1 else "SHORT",side,round(e,6),round(e*(1+tp*side),6),round(e*(1-sl*side),6),
                     "" if hit is None else ("TP" if hit else "SL"),round(fwd,4)])
    rows.sort()
    w=csv.writer(open("02_directional_calls.csv","w",newline=""))
    w.writerow(["timestamp","asset","side","prediction","entry","take_profit","stop_loss","first_hit","outcome"]); w.writerows(rows)
    print("02 claimed win rate (TP before SL):", round(wins/len(rows)*100,1), "%")

# ---------- 03 daily probability forecast, single series, 1d ----------
def f03():
    n=730; t0=datetime(2024,9,1)
    px=[30000.0]; p_up=[]
    rows=[]
    for i in range(n):
        latent=R.gauss(0,1)
        prob=1/(1+math.exp(-0.9*latent))
        ret=0.0006+0.004*latent*0.0+R.gauss(0,0.03)  # no edge at all
        rows.append([(t0+timedelta(days=i)).strftime("%Y-%m-%d"),round(prob,3),round(px[-1],2),None])
        px.append(px[-1]*(1+ret))
    for i,r in enumerate(rows):
        r[3]=round((px[i+1]/px[i]-1)*100,4)
    w=csv.writer(open("03_daily_probability_forecast.csv","w",newline=""))
    w.writerow(["date","prob_up","price","next_day_return"]); w.writerows(rows)

# ---------- 04 weekly demand forecast per store (non-crypto), genuine edge ----------
def f04():
    stores=[f"STORE_{i:02d}" for i in range(1,13)]; t0=datetime(2024,1,1); weeks=104
    rows=[]
    for s in stores:
        level=R.uniform(800,3000)
        for k in range(weeks):
            true_g=R.gauss(0,6)+ (8*math.sin(2*math.pi*k/52))  # seasonality, %
            fc=0.7*true_g+R.gauss(0,3)            # forecaster sees most of it
            actual=true_g+R.gauss(0,4)
            rows.append([(t0+timedelta(weeks=k)).strftime("%Y-%m-%d"),s,round(fc,2),round(actual,2),round(level,1)])
            level*=1+actual/100
    rows.sort()
    w=csv.writer(open("04_weekly_demand_forecast.csv","w",newline=""))
    w.writerow(["timestamp","group","prediction","outcome","units"]); w.writerows(rows)

# ---------- 05 common mistakes: Danish Excel export, duplicates, missing outcomes ----------
def f05():
    t0=datetime(2026,6,1); n=24*90
    px=walk(n+4,0.004,start=2500)
    L=[]
    for i in range(n):
        ts=(t0+timedelta(hours=i)).strftime("%d-%m-%Y %H:%M")
        pred=R.gauss(0,1)
        out=(px[i+4]/px[i]-1)*100 if i < n-40 else None   # the last 40 rows have no outcome yet
        row=[ts, f"{pred:.3f}".replace(".",","), "" if out is None else f"{out:.4f}".replace(".",","), f"{px[i]:.2f}".replace(".",",")]
        L.append(row)
        if i%3==0: L.append(list(row))   # duplicate: the same timestamp twice
    w=csv.writer(open("05_common_mistakes_excel_dk.csv","w",newline="",encoding="utf-8-sig"),delimiter=";")
    w.writerow(["Dato","Forudsigelse","Udfald","Kurs"]); w.writerows(L)

# ---------- 06 lookahead bug: outcome is the PAST 4h move ----------
def f06():
    assets=["BTC","ETH","SOL"]; t0=datetime(2026,4,1); n=4*24*45
    rows=[]
    for a in assets:
        px=walk(n+16,0.003,start=R.uniform(10,1000))
        for i in range(16,n):
            past=(px[i]/px[i-16]-1)*100
            pred=50+10*past/1.0+R.gauss(0,4)     # model uses recent move (fine)
            rows.append([(t0+timedelta(minutes=15*i)).strftime("%Y-%m-%d %H:%M"),a,round(pred,2),round(px[i],4),round(past,4)])  # BUG: outcome = past move
    rows.sort()
    w=csv.writer(open("06_misaligned_outcome.csv","w",newline=""))
    w.writerow(["timestamp","asset","signal","price","fwd_return_4h"]); w.writerows(rows)

if __name__ == "__main__":
    out = sys.argv[sys.argv.index("--out") + 1] if "--out" in sys.argv else os.path.join(os.path.dirname(os.path.abspath(__file__)), "data")
    os.makedirs(out, exist_ok=True)
    os.chdir(out)
    for f in (f01, f02, f03, f04, f05, f06):
        f()
    print("wrote 6 files to", out)