45 lines
2.1 KiB
Python
45 lines
2.1 KiB
Python
import numpy as np
|
|
import pandas as pd
|
|
|
|
def load_csv(path: str) -> pd.DataFrame:
|
|
"""CSV exporté de MT5 (Historique) ou générique: time/date, open, high, low, close."""
|
|
df = pd.read_csv(path, sep=None, engine="python")
|
|
df.columns = [c.strip().strip("<>").lower() for c in df.columns]
|
|
if "time" in df.columns and "date" in df.columns:
|
|
df["time"] = pd.to_datetime(df["date"].astype(str) + " " + df["time"].astype(str))
|
|
elif "date" in df.columns:
|
|
df["time"] = pd.to_datetime(df["date"])
|
|
else:
|
|
df["time"] = pd.to_datetime(df["time"])
|
|
df = df[["time", "open", "high", "low", "close"]].sort_values("time").drop_duplicates("time")
|
|
return df.reset_index(drop=True)
|
|
|
|
def from_mt5(symbol: str, tf: str, bars: int, cfg_account: dict) -> pd.DataFrame:
|
|
"""Télécharge l'historique depuis un terminal MT5 (Windows)."""
|
|
from core.connector import MT5
|
|
mt = MT5(cfg_account, 0, 0)
|
|
if not mt.connect():
|
|
raise SystemExit("Connexion MT5 impossible")
|
|
df = mt.rates(symbol, tf, bars)
|
|
mt.shutdown()
|
|
return df[["time", "open", "high", "low", "close"]].reset_index(drop=True)
|
|
|
|
def synthetic(bars: int = 12000, start_price: float = 2000.0, seed: int = 7, freq: str = "15min") -> pd.DataFrame:
|
|
"""Données FICTIVES (régimes tendance/range). Sert uniquement à tester le moteur."""
|
|
rng = np.random.default_rng(seed)
|
|
vol = np.zeros(bars); drift = np.zeros(bars)
|
|
v, d = 0.6, 0.0
|
|
for i in range(bars):
|
|
if i % 400 == 0:
|
|
d = rng.choice([-0.05, 0.0, 0.0, 0.05]) * start_price / 2000
|
|
v = rng.uniform(0.4, 1.2) * start_price / 2000
|
|
vol[i], drift[i] = v, d
|
|
ret = drift + vol * rng.standard_normal(bars)
|
|
close = start_price + np.cumsum(ret)
|
|
open_ = np.concatenate([[start_price], close[:-1]])
|
|
wick = np.abs(rng.standard_normal(bars)) * vol * 0.6
|
|
high = np.maximum(open_, close) + wick
|
|
low = np.minimum(open_, close) - np.abs(rng.standard_normal(bars)) * vol * 0.6
|
|
t = pd.date_range("2025-01-06", periods=bars * 2, freq=freq)
|
|
t = t[t.dayofweek < 5][:bars]
|
|
return pd.DataFrame({"time": t, "open": open_, "high": high, "low": low, "close": close})
|