# _05_TradingSection/_02_Strategy/CreateStrategies.py # ============================================================ # STRATEGY CREATION # Rebuilds every trigger straight on the candle arrays, backtests # the whole grid on all three assets, and writes each combination # that clears the benchmark on both brokers as its own strategy. # ============================================================ import os # filesystem paths and folder creation import glob # market data file lookup import json # candle parsing import numpy as np # vectorized engine from numpy.lib.stride_tricks import sliding_window_view # rolling windows for the indicators # ============================================================ # ASSETS (created directly on all three) # ============================================================ ASSETS = ["NAS100", "US30", "SP500"] # every asset generated in one run BROKERS = ["alpari", "acg"] # a strategy has to hold on both brokers DATA_ROOT = r"C:\_01_Marketdata\_02_MT5" # root of the MT5 market data OUT_ROOT = r"C:\_02_LiveStrategies" # where the approved strategies are written # ============================================================ # PARAMETER GRID # ============================================================ EXEC_TFS = ["M1", "M2"] # execution timeframes MA_MODES = ["Simple", "Exponential", "Smoothed", "LinearWeighted"] # every moving average mode of MT5 MA_PAIRS = [(3, 10), (5, 20), (8, 34), (9, 21), (20, 50), (50, 200)] # fast / slow moving average pairs SL_LIST = [10, 20, 30, 40] # stop loss distances in points RR_LIST = [None, 1.5, 2.0, 3.0] # take profit as a multiple of the stop loss DUR_LIST = [None, 2, 3, 5, 10] # maximum life of the trade in candles # ============================================================ # BENCHMARK (checked on each broker, never the kinder one) # ============================================================ START_BAL = 100000.0 # starting balance of the account MIN_NET_PNL = 500000.0 # net profit required over the tested period MIN_TRADES_PER_WEEK = 100.0 # executions per week required H = 240 # candles looked ahead to resolve a trade # ============================================================ # DATA LOADING (one json per asset, broker and timeframe) # ============================================================ def load(asset, broker, tf): fp = glob.glob(os.path.join(DATA_ROOT, broker, asset, "*_" + tf, "*.json"))[0] # resolve the real file on disk rows = json.load(open(fp, encoding="utf-8")) # the candles of one asset, one timeframe o = np.array([r["open"] for r in rows], np.float64) # open prices h = np.array([r["high"] for r in rows], np.float64) # high prices l = np.array([r["low"] for r in rows], np.float64) # low prices c = np.array([r["close"] for r in rows], np.float64) # close prices return {"o": o, "h": h, "l": l, "c": c} # ============================================================ # MOVING AVERAGES (faithful to MT5) # ============================================================ def sma(x, n): cs = np.cumsum(np.insert(x, 0, 0.0)); out = np.full(len(x), np.nan) out[n - 1:] = (cs[n:] - cs[:-n]) / n # simple moving average return out def lwma(x, n): w = np.arange(1, n + 1, dtype=np.float64); out = np.full(len(x), np.nan) out[n - 1:] = (sliding_window_view(x, n) * w).sum(1) / w.sum() # linear weighted moving average return out def ema(x, n): out = np.empty(len(x)); pr = 2.0 / (n + 1.0); out[0] = x[0] for i in range(1, len(x)): out[i] = x[i] * pr + out[i - 1] * (1.0 - pr) # exponential moving average return out def moving_average(x, n, mode): if mode == "Exponential": return ema(x, n) if mode == "LinearWeighted": return lwma(x, n) return sma(x, n) # simple and smoothed share the base here # ============================================================ # TRIGGER (moving average cross, rebuilt on the whole array at once) # ============================================================ def moving_average_cross(c, mode, fast, slow): d = moving_average(c, fast, mode) - moving_average(c, slow, mode) # distance between the two averages s = np.zeros(len(c), np.int8) s[1:][(d[1:] > 0) & (d[:-1] <= 0)] = 1 # fast crossing above the slow -> long s[1:][(d[1:] < 0) & (d[:-1] >= 0)] = -1 # fast crossing below the slow -> short return s # ============================================================ # OUTCOME (stop loss, take profit and maximum life in one vectorized pass) # ============================================================ def backtest(candles, signal, sl, rr, dur): o, h, l, c = candles["o"], candles["h"], candles["l"], candles["c"]; N = len(c) ent = np.nonzero(signal != 0)[0] + 1; ent = ent[ent < N] # entry on the candle after the trigger d = signal[ent - 1]; ep = o[ent] # direction and entry price tp = sl * rr if rr is not None else np.inf # take profit distance, infinite when there is none cap = min(dur if dur is not None else H, H) # candles the trade can stay open off = np.arange(H) rows = np.clip(ent[:, None] + off[None, :], 0, N - 1) # window of candles ahead of every entry buy = (d == 1)[:, None] adverse = np.where(buy, ep[:, None] - np.minimum.accumulate(l[rows], 1), np.maximum.accumulate(h[rows], 1) - ep[:, None]) # worst move against the trade favour = np.where(buy, np.maximum.accumulate(h[rows], 1) - ep[:, None], ep[:, None] - np.minimum.accumulate(l[rows], 1)) # best move in favour of the trade hit_sl = (adverse < sl).sum(1); hit_tp = (favour < tp).sum(1) # first candle that touches each level sl_first = (hit_sl <= hit_tp) & (hit_sl < cap) pnl_pts = np.where(sl_first, -float(sl), np.where(hit_tp < cap, tp, 0.0)) # result in points, capped trades flat return pnl_pts, ent # ============================================================ # WALK (sequential balance with dynamic risk, no overlapping trades) # ============================================================ def walk(pnl_pts, ent, sl): bal = START_BAL; last = -1; balances = [] for i in range(len(ent)): if ent[i] <= last: # the previous trade is still open continue risk = min(3500.0, max(100.0, (bal - 90000.0) / 20.0)) # dynamic risk on the current balance bal += pnl_pts[i] / sl * risk balances.append(bal); last = ent[i] + 1 return balances # ============================================================ # BENCHMARK # ============================================================ def passes(balances, weeks): if not balances: return False net = balances[-1] - START_BAL # net profit over the period per_week = len(balances) / weeks # executions per week return net >= MIN_NET_PNL and per_week >= MIN_TRADES_PER_WEEK # ============================================================ # MAIN (loop the grid on every asset, write what survives) # ============================================================ def main(): for asset in ASSETS: # create directly on all three assets data = {b: {tf: load(asset, b, tf) for tf in EXEC_TFS} for b in BROKERS} weeks = {b: len(data[b]["M1"]["c"]) / (7 * 24 * 60) for b in BROKERS} # length of the test period in weeks kept = 0 for tf in EXEC_TFS: for mode in MA_MODES: for fast, slow in MA_PAIRS: signal = {b: moving_average_cross(data[b][tf]["c"], mode, fast, slow) for b in BROKERS} for sl in SL_LIST: for rr in RR_LIST: for dur in DUR_LIST: ok = True for b in BROKERS: # both brokers have to clear the benchmark pnl_pts, ent = backtest(data[b][tf], signal[b], sl, rr, dur) if not passes(walk(pnl_pts, ent, sl), weeks[b]): ok = False break if ok: name = f"{asset}_{tf}_MovingAverageCross{mode}-{fast}-{slow}_sl{sl}_tp{rr}_dur{dur}" os.makedirs(os.path.join(OUT_ROOT, asset, name), exist_ok=True) # one folder per strategy kept += 1 print(f"{asset}: {kept} strategies written") if __name__ == "__main__": main()