| """
|
| Multi-symbol strategy screen, ranked the way the account is actually judged.
|
|
|
| RANKING RULE (non-negotiable here): drawdown first, not profit. A candidate is
|
| only reported as viable if it clears ALL THREE of
|
| * cadence >= 2 trades / month (a strategy that never trades is not one)
|
| * maxDD <= the prop limit
|
| * ret/DD >= 2
|
| Profit ranking is what produces the curve-fit winners this project keeps
|
| re-discovering, so it is deliberately not the sort key.
|
|
|
| COSTS ARE REAL. The exported `spread` column is in POINTS at each bar, and
|
| point size is recovered from the decimal count the exporter wrote (it used
|
| DoubleToString(price, digits), so the file's precision IS the symbol's digits).
|
| A long round trip pays the full spread once: you buy the ask and sell the bid.
|
| This matters more than any parameter -- the decade verdict on this project was
|
| that spread is the whole gap between paper and traded results.
|
|
|
| CONTROL. Every candidate is compared against a LONG-ONLY RANDOM ENTRY with a
|
| matched holding period, not a two-sided coin flip. On a drifting index a
|
| two-sided control is a straw man: it loses money by construction, so beating
|
| it proves nothing. Long bias on an index is not a flaw, it is the asset.
|
| """
|
|
|
| from __future__ import annotations
|
|
|
| import numpy as np
|
|
|
| COMMON = r"C:\Users\admin\AppData\Roaming\MetaQuotes\Terminal\Common\Files"
|
|
|
|
|
| # ------------------------------------------------------------------ loading
|
| def load(symbol: str, period: str = "PERIOD_D1"):
|
| path = rf"{COMMON}\bars_{symbol}_{period}.csv"
|
| raw = np.genfromtxt(path, delimiter=",", skip_header=1, dtype=str, encoding="ansi")
|
| ts = np.array([f"{r[0][:10].replace('.', '-')}T{r[0][11:]}" for r in raw],
|
| dtype="datetime64[s]")
|
| o, h, l, c = (raw[:, i].astype(float) for i in (1, 2, 3, 4))
|
| vol = raw[:, 5].astype(float)
|
| spread_pts = raw[:, 6].astype(float)
|
| # digits == decimals the exporter wrote; point == 10^-digits
|
| digits = max(len(s.split(".")[1]) if "." in s else 0 for s in raw[:20, 4])
|
| point = 10.0 ** (-digits)
|
| return dict(ts=ts, o=o, h=h, l=l, c=c, v=vol,
|
| cost=spread_pts * point, point=point, digits=digits, symbol=symbol)
|
|
|
|
|
| # --------------------------------------------------------------- indicators
|
| def sma(x, n):
|
| out = np.full(len(x), np.nan)
|
| cs = np.concatenate([[0.0], np.cumsum(x)])
|
| out[n - 1:] = (cs[n:] - cs[:-n]) / n
|
| return out
|
|
|
|
|
| def rolling_std(x, n):
|
| out = np.full(len(x), np.nan)
|
| cs = np.concatenate([[0.0], np.cumsum(x)])
|
| cs2 = np.concatenate([[0.0], np.cumsum(x * x)])
|
| m = (cs[n:] - cs[:-n]) / n
|
| m2 = (cs2[n:] - cs2[:-n]) / n
|
| out[n - 1:] = np.sqrt(np.maximum(m2 - m * m, 0.0))
|
| return out
|
|
|
|
|
| def atr(h, l, c, n=14):
|
| tr = np.maximum(h[1:] - l[1:], np.maximum(np.abs(h[1:] - c[:-1]), np.abs(l[1:] - c[:-1])))
|
| tr = np.concatenate([[h[0] - l[0]], tr])
|
| out = np.full(len(tr), np.nan)
|
| if len(tr) <= n:
|
| return out
|
| out[n - 1] = tr[:n].mean()
|
| for i in range(n, len(tr)):
|
| out[i] = (out[i - 1] * (n - 1) + tr[i]) / n
|
| return out
|
|
|
|
|
| def rsi(c, n=2):
|
| d = np.diff(c, prepend=c[0])
|
| up = np.where(d > 0, d, 0.0)
|
| dn = np.where(d < 0, -d, 0.0)
|
| au = np.full(len(c), np.nan)
|
| ad = np.full(len(c), np.nan)
|
| if len(c) <= n:
|
| return au
|
| au[n] = up[1:n + 1].mean()
|
| ad[n] = dn[1:n + 1].mean()
|
| for i in range(n + 1, len(c)):
|
| au[i] = (au[i - 1] * (n - 1) + up[i]) / n
|
| ad[i] = (ad[i - 1] * (n - 1) + dn[i]) / n
|
| rs = au / np.where(ad > 0, ad, np.nan)
|
| out = 100 - 100 / (1 + rs)
|
| out[ad == 0] = 100.0
|
| return out
|
|
|
|
|
| # ----------------------------------------------------------------- simulate
|
| def simulate(d, entries, side=1, exit_ma=None, max_bars=10, stop_atr=3.0, atr_n=14):
|
| """Bar-close signals, NEXT-BAR-OPEN fills. Returns per-trade log.
|
|
|
| entries : bool array; entry decided on bar i, filled at open of i+1.
|
| exit_ma : if given, exit when close crosses back to this level; else time/stop.
|
|
|
| NO LOOKAHEAD: every exit test uses bar j's own high/low/close, and the stop
|
| is checked against the bar's extreme before its close, so an intrabar stop
|
| is honoured rather than assumed away.
|
| """
|
| o, h, l, c = d["o"], d["h"], d["l"], d["c"]
|
| a = atr(h, l, c, atr_n)
|
| cost = d["cost"]
|
| ts = d["ts"]
|
| trades = []
|
| i = 0
|
| n = len(c)
|
| while i < n - 2:
|
| if not entries[i] or not np.isfinite(a[i]) or a[i] <= 0:
|
| i += 1
|
| continue
|
| fill = i + 1
|
| entry = o[fill]
|
| if not np.isfinite(entry) or entry <= 0:
|
| i += 1
|
| continue
|
| stop = entry - side * stop_atr * a[i]
|
| exit_px, exit_j, reason = None, None, None
|
| for j in range(fill, min(fill + max_bars, n)):
|
| # stop first: within a bar we cannot know order, so assume the worse
|
| if side > 0 and l[j] <= stop:
|
| exit_px, exit_j, reason = stop, j, "stop"
|
| break
|
| if side < 0 and h[j] >= stop:
|
| exit_px, exit_j, reason = stop, j, "stop"
|
| break
|
| if exit_ma is not None and np.isfinite(exit_ma[j]):
|
| if (side > 0 and c[j] >= exit_ma[j]) or (side < 0 and c[j] <= exit_ma[j]):
|
| exit_px, exit_j, reason = c[j], j, "target"
|
| break
|
| if exit_px is None:
|
| exit_j = min(fill + max_bars - 1, n - 1)
|
| exit_px, reason = c[exit_j], "time"
|
| gross = side * (exit_px - entry)
|
| net = gross - cost[fill] # full spread once, round trip
|
| trades.append(dict(entry_i=fill, exit_i=exit_j, t=ts[fill],
|
| ret=net / entry, gross=gross / entry, reason=reason,
|
| bars=exit_j - fill + 1))
|
| i = exit_j + 1 # no overlapping positions
|
| return trades
|
|
|
|
|
| def metrics(trades, ts, risk_frac=0.01, stop_atr=3.0):
|
| """Equity curve at fixed fractional risk; drawdown on that curve.
|
|
|
| Each trade risks `risk_frac` of equity at the stop distance, so a trade's
|
| equity impact is risk_frac * (net return / stop distance) -- i.e. R
|
| multiples, which is how the prop limit is actually consumed.
|
| """
|
| if not trades:
|
| return None
|
| years = (ts[-1] - ts[0]) / np.timedelta64(365, "D")
|
| months = max(years * 12.0, 1e-9)
|
| rets = np.array([t["ret"] for t in trades])
|
| eq = [1.0]
|
| for t in trades:
|
| # R multiple: net move relative to the risked distance
|
| eq.append(eq[-1] * (1.0 + risk_frac * t["r"]))
|
| eq = np.array(eq)
|
| peak = np.maximum.accumulate(eq)
|
| dd = (peak - eq) / peak
|
| total = eq[-1] - 1.0
|
| maxdd = dd.max()
|
| return dict(n=len(trades), per_month=len(trades) / months,
|
| exp_bp=rets.mean() * 1e4, hit=float((rets > 0).mean()),
|
| total=total, maxdd=maxdd,
|
| ret_dd=(total / maxdd if maxdd > 1e-9 else np.nan),
|
| cagr=(eq[-1] ** (1 / max(years, 1e-9)) - 1.0),
|
| years=years)
|
|
|
|
|
| def add_r(trades, d, stop_atr=3.0, atr_n=14):
|
| """Attach the R multiple to each trade (net return / risked distance)."""
|
| a = atr(d["h"], d["l"], d["c"], atr_n)
|
| for t in trades:
|
| i = t["entry_i"] - 1
|
| risk = stop_atr * a[i] / d["o"][t["entry_i"]]
|
| t["r"] = t["ret"] / risk if risk > 0 else 0.0
|
| return trades
|
ff46cfd57e