186 行
7.2 KiB
Python
186 行
7.2 KiB
Python
 | """
| ||
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
|