File size: 4,710 Bytes
7c3e5dd c3c03ee 7c3e5dd c3c03ee 7c3e5dd c3c03ee 7c3e5dd c3c03ee 7c3e5dd c3c03ee 7c3e5dd c3c03ee 7c3e5dd c3c03ee 7c3e5dd c3c03ee 7c3e5dd c3c03ee 7c3e5dd c3c03ee 7c3e5dd c3c03ee 7c3e5dd c3c03ee 7c3e5dd c3c03ee 7c3e5dd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | # /// script
# dependencies = ["torch", "transformers", "accelerate", "pandas", "numpy", "datasets", "tqdm"]
# ///
import os
import json
import pandas as pd
import numpy as np
import torch
from transformers import AutoTokenizer, AutoModelForSeq2SeqLM
from datasets import load_dataset
from tqdm import tqdm
# --- CONFIG ---
MODEL_ID = "AbdelrehmanFouad/t5-efficient-base-usdjpy-forecaster"
SPREAD = 0.015 # 1.5 pips average spread for USDJPY
HORIZON = 24 # hours
def load_backtest_data():
print("Loading data for backtest...")
ds_jpy = load_dataset("huggingXG/forex_USDJPY", split="train", streaming=True)
data = []
# Collect enough to have history + test slice
for i, row in enumerate(ds_jpy):
data.append(row)
if i >= 3000000: break
df = pd.DataFrame(data)
df['timestamp'] = pd.to_datetime(df['timestamp'], format='ISO8601')
df.set_index('timestamp', inplace=True)
df['mid'] = (df['ask'] + df['bid']) / 2
resampled = df['mid'].resample('1h').ohlc()
resampled.dropna(inplace=True)
ds_cal = load_dataset("Ehsanrs2/Forex_Factory_Calendar", split="train")
df_cal = ds_cal.to_pandas()
df_cal['DateTime'] = pd.to_datetime(df_cal['DateTime'], utc=True)
split_idx = int(len(resampled) * 0.8)
test_resampled = resampled.iloc[split_idx - 168:]
return test_resampled, df_cal
def run_backtest():
test_data, df_cal = load_backtest_data()
print("Loading model...")
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
model = AutoModelForSeq2SeqLM.from_pretrained(MODEL_ID, device_map="auto")
model.eval()
context_size = 168
results = []
if test_data.index.tz is None:
test_data.index = test_data.index.tz_localize('UTC')
indices = range(context_size, len(test_data) - HORIZON, 12) # Step 12 for speed
print(f"Running backtest over {len(indices)} trades...")
for i in tqdm(indices):
current_time = test_data.index[i]
context_window = test_data.iloc[i-context_size:i]['close'].tolist()
entry_price = test_data.iloc[i]['close']
actual_exit_price = test_data.iloc[i+HORIZON]['close']
# Calendar events
end_time = test_data.index[i+HORIZON-1]
events = df_cal[(df_cal['DateTime'] >= current_time) & (df_cal['DateTime'] <= end_time)]
macro_events = []
for _, event in events.iterrows():
macro_events.append({
"time": event['DateTime'].strftime("%Y-%m-%dT%H:%M:%SZ"),
"event": event['Event'],
"impact": event['Impact'].lower().split()[0]
})
input_json = {"context_window": [round(float(c), 4) for c in context_window], "macro_events": macro_events}
prompt = "forecast usdjpy: " + json.dumps(input_json)
inputs = tokenizer(prompt, return_tensors="pt", max_length=1024, truncation=True).to(model.device)
with torch.no_grad():
outputs = model.generate(**inputs, max_new_tokens=512)
raw_pred = tokenizer.decode(outputs[0], skip_special_tokens=True)
# Robust parsing
direction = "sideways"
try:
parse_target = raw_pred.strip()
if not parse_target.startswith("{"): parse_target = "{" + parse_target + "}"
pred_data = json.loads(parse_target)
direction = pred_data.get("direction", "sideways")
except:
if '"direction": "up"' in raw_pred: direction = "up"
elif '"direction": "down"' in raw_pred: direction = "down"
pnl = 0
if direction == "up":
pnl = (actual_exit_price - (entry_price + SPREAD))
elif direction == "down":
pnl = ((entry_price - SPREAD) - actual_exit_price)
results.append({
"time": str(current_time),
"direction": direction,
"pnl": float(pnl),
"win": (pnl > 0) if direction != "sideways" else None
})
df_results = pd.DataFrame(results)
trades = df_results[df_results['direction'] != 'sideways']
total_pips = df_results['pnl'].sum()
win_rate = trades['win'].mean() if len(trades) > 0 else 0
stats = {
"total_periods": len(df_results),
"total_trades": len(trades),
"win_rate": f"{win_rate:.2%}",
"net_pips": f"{total_pips:.4f}",
"avg_pnl": f"{(total_pips/len(trades)):.4f}" if len(trades) > 0 else "0"
}
print("\n--- RESULTS ---")
print(json.dumps(stats, indent=2))
print("----------------")
if __name__ == "__main__":
run_backtest()
|