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