docs(examples): add 3 end-to-end strategy examples (Rust + Python) (#65)
Wires real indicators into complete signal -> fill -> PnL -> equity loops over the checked-in BTCUSDT datasets, with per-trade Sharpe and max-drawdown reported on stdout. Closes the gap where existing examples showed only the mechanics of calling `update`/`batch` but not how Wickra plugs into a trading-system shape. Three strategies, each in Rust + Python (six files total): - strategy_rsi_mean_reversion — RSI(14) thresholds (30/70) on 1h BTCUSDT. Binary position, 0.1% per-trade fee. - strategy_macd_adx — MACD crossover entries gated by ADX(14) > 20 on 1h BTCUSDT. Trend-follower demo of multi-indicator gating. - strategy_bollinger_squeeze — Bollinger-bandwidth 180-day-low squeeze + upper-band breakout entry, ATR(14) * 2 stop. On 1d BTCUSDT for interpretable lookback. Each file is self-contained — print_summary is inlined per script so the example stays a single-file read. Every script prints a NOT-financial-advice notice next to its results. examples/README.md updated to list the new bins/scripts.
This commit is contained in:
kingchenc
@@ -0,0 +1,178 @@
|
||||
"""Strategy example: Bollinger-Squeeze breakout with ATR-based stop.
|
||||
|
||||
Enters long when the Bollinger Bandwidth has just printed a fresh
|
||||
6-month low (the squeeze) and price closes above the upper band (the
|
||||
release). Exits when price closes below entry minus 2 * ATR(14), or
|
||||
when the upper band rolls back under the entry price. 0.1% fees per
|
||||
trade.
|
||||
|
||||
Educational example. NOT a live trading recommendation.
|
||||
|
||||
Run with::
|
||||
|
||||
python -m examples.python.strategy_bollinger_squeeze
|
||||
|
||||
Uses the checked-in ``examples/data/btcusdt-1d.csv`` dataset because
|
||||
daily bars give an interpretable 6-month-low lookback (~180 bars).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import math
|
||||
from collections import deque
|
||||
from pathlib import Path
|
||||
|
||||
import wickra as ta
|
||||
|
||||
FEE = 0.001
|
||||
BB_PERIOD = 20
|
||||
BB_K = 2.0
|
||||
ATR_PERIOD = 14
|
||||
ATR_STOP_MULT = 2.0
|
||||
SQUEEZE_LOOKBACK = 180
|
||||
|
||||
|
||||
def load_candles(path: Path) -> list[dict[str, float]]:
|
||||
with path.open() as fh:
|
||||
reader = csv.DictReader(fh)
|
||||
return [
|
||||
{
|
||||
"open": float(r["open"]),
|
||||
"high": float(r["high"]),
|
||||
"low": float(r["low"]),
|
||||
"close": float(r["close"]),
|
||||
"volume": float(r["volume"]),
|
||||
}
|
||||
for r in reader
|
||||
]
|
||||
|
||||
|
||||
def print_summary(
|
||||
name: str,
|
||||
first_price: float,
|
||||
last_price: float,
|
||||
bars: int,
|
||||
closed_trades: list[float],
|
||||
final_equity: float,
|
||||
equity_curve: list[float],
|
||||
) -> None:
|
||||
buy_hold = last_price / first_price
|
||||
strat_return = final_equity - 1.0
|
||||
bh_return = buy_hold - 1.0
|
||||
wins = sum(1 for r in closed_trades if r > 0)
|
||||
losses = sum(1 for r in closed_trades if r < 0)
|
||||
best = max(closed_trades) if closed_trades else 0.0
|
||||
worst = min(closed_trades) if closed_trades else 0.0
|
||||
n = len(closed_trades)
|
||||
mean_ret = sum(closed_trades) / n if n else 0.0
|
||||
var_ret = (
|
||||
sum((r - mean_ret) ** 2 for r in closed_trades) / (n - 1) if n > 1 else 0.0
|
||||
)
|
||||
sharpe = mean_ret / math.sqrt(var_ret) if var_ret > 0 else 0.0
|
||||
peak = equity_curve[0] if equity_curve else 1.0
|
||||
max_dd = 0.0
|
||||
for eq in equity_curve:
|
||||
if eq > peak:
|
||||
peak = eq
|
||||
dd = (peak - eq) / peak
|
||||
if dd > max_dd:
|
||||
max_dd = dd
|
||||
print(f"=== {name} ===")
|
||||
print(f"Bars: {bars}")
|
||||
print(f"Trades: {n} (W{wins} / L{losses})")
|
||||
print(f"Strategy return: {strat_return * 100:+.2f}%")
|
||||
print(f"Buy & Hold return: {bh_return * 100:+.2f}%")
|
||||
print(f"Excess over BH: {(strat_return - bh_return) * 100:+.2f}%")
|
||||
print(f"Max drawdown: {max_dd * 100:.2f}%")
|
||||
print(
|
||||
f"Per-trade Sharpe: {sharpe:.2f} "
|
||||
f"(mean {mean_ret:+.4f}, stddev {math.sqrt(var_ret):.4f})"
|
||||
)
|
||||
print(f"Best / worst trade: {best * 100:+.2f}% / {worst * 100:+.2f}%")
|
||||
print()
|
||||
print(
|
||||
"NOTE: Educational example — fees, slippage, funding costs and tax "
|
||||
"effects are simplified or omitted. Past performance is not "
|
||||
"indicative of future results."
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
path = Path(__file__).resolve().parents[1] / "data" / "btcusdt-1d.csv"
|
||||
candles = load_candles(path)
|
||||
if len(candles) < SQUEEZE_LOOKBACK + BB_PERIOD:
|
||||
raise SystemExit(
|
||||
f"dataset has only {len(candles)} bars; need at least "
|
||||
f"{SQUEEZE_LOOKBACK + BB_PERIOD}"
|
||||
)
|
||||
|
||||
bb = ta.BollingerBands(BB_PERIOD, BB_K)
|
||||
atr = ta.ATR(ATR_PERIOD)
|
||||
bw_window: deque[float] = deque(maxlen=SQUEEZE_LOOKBACK)
|
||||
|
||||
in_position = False
|
||||
entry_price = 0.0
|
||||
stop_level = 0.0
|
||||
closed_trades: list[float] = []
|
||||
equity = 1.0
|
||||
equity_curve: list[float] = []
|
||||
|
||||
for c in candles:
|
||||
bb_out = bb.update(c["close"])
|
||||
atr_val = atr.update(c["high"], c["low"], c["close"])
|
||||
price = c["close"]
|
||||
mtm = equity * (price / entry_price) if in_position else equity
|
||||
equity_curve.append(mtm)
|
||||
|
||||
if bb_out is None or atr_val is None:
|
||||
continue
|
||||
|
||||
upper = bb_out[0] if isinstance(bb_out, tuple) else bb_out.upper
|
||||
middle = bb_out[1] if isinstance(bb_out, tuple) else bb_out.middle
|
||||
lower = bb_out[2] if isinstance(bb_out, tuple) else bb_out.lower
|
||||
bandwidth = (upper - lower) / middle if abs(middle) > 1e-12 else float("nan")
|
||||
|
||||
if math.isnan(bandwidth):
|
||||
continue
|
||||
bw_window.append(bandwidth)
|
||||
if len(bw_window) < SQUEEZE_LOOKBACK:
|
||||
continue
|
||||
min_bw = min(bw_window)
|
||||
|
||||
if in_position:
|
||||
stop_hit = price < stop_level
|
||||
upper_collapse = upper < entry_price
|
||||
if stop_hit or upper_collapse:
|
||||
trade_ret = price / entry_price - 1.0
|
||||
closed_trades.append(trade_ret)
|
||||
equity *= (1.0 + trade_ret) * (1.0 - FEE)
|
||||
in_position = False
|
||||
else:
|
||||
is_new_low = abs(bandwidth - min_bw) < 1e-12
|
||||
breakout = price > upper
|
||||
if is_new_low and breakout:
|
||||
entry_price = price
|
||||
stop_level = price - ATR_STOP_MULT * atr_val
|
||||
equity *= 1.0 - FEE
|
||||
in_position = True
|
||||
|
||||
if in_position:
|
||||
last_price = candles[-1]["close"]
|
||||
trade_ret = last_price / entry_price - 1.0
|
||||
closed_trades.append(trade_ret)
|
||||
equity *= (1.0 + trade_ret) * (1.0 - FEE)
|
||||
|
||||
print_summary(
|
||||
"Bollinger Squeeze Breakout (1d, BTCUSDT)",
|
||||
candles[0]["close"],
|
||||
candles[-1]["close"],
|
||||
len(candles),
|
||||
closed_trades,
|
||||
equity,
|
||||
equity_curve,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,155 @@
|
||||
"""Strategy example: MACD crossover with ADX trend-strength filter.
|
||||
|
||||
Long-only trend follower. Entries fire on a MACD-line-crosses-above-
|
||||
signal-line event while ADX(14) > 20 (i.e. directional market). Exits
|
||||
on the opposite MACD crossover regardless of ADX. 0.1% fees per trade.
|
||||
|
||||
Educational example. NOT a live trading recommendation.
|
||||
|
||||
Run with::
|
||||
|
||||
python -m examples.python.strategy_macd_adx
|
||||
|
||||
Uses the checked-in ``examples/data/btcusdt-1h.csv`` dataset.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import math
|
||||
from pathlib import Path
|
||||
|
||||
import wickra as ta
|
||||
|
||||
FEE = 0.001
|
||||
ADX_FLOOR = 20.0
|
||||
|
||||
|
||||
def load_candles(path: Path) -> list[dict[str, float]]:
|
||||
with path.open() as fh:
|
||||
reader = csv.DictReader(fh)
|
||||
return [
|
||||
{
|
||||
"open": float(r["open"]),
|
||||
"high": float(r["high"]),
|
||||
"low": float(r["low"]),
|
||||
"close": float(r["close"]),
|
||||
"volume": float(r["volume"]),
|
||||
}
|
||||
for r in reader
|
||||
]
|
||||
|
||||
|
||||
def print_summary(
|
||||
name: str,
|
||||
first_price: float,
|
||||
last_price: float,
|
||||
bars: int,
|
||||
closed_trades: list[float],
|
||||
final_equity: float,
|
||||
equity_curve: list[float],
|
||||
) -> None:
|
||||
buy_hold = last_price / first_price
|
||||
strat_return = final_equity - 1.0
|
||||
bh_return = buy_hold - 1.0
|
||||
wins = sum(1 for r in closed_trades if r > 0)
|
||||
losses = sum(1 for r in closed_trades if r < 0)
|
||||
best = max(closed_trades) if closed_trades else 0.0
|
||||
worst = min(closed_trades) if closed_trades else 0.0
|
||||
n = len(closed_trades)
|
||||
mean_ret = sum(closed_trades) / n if n else 0.0
|
||||
var_ret = (
|
||||
sum((r - mean_ret) ** 2 for r in closed_trades) / (n - 1) if n > 1 else 0.0
|
||||
)
|
||||
sharpe = mean_ret / math.sqrt(var_ret) if var_ret > 0 else 0.0
|
||||
peak = equity_curve[0] if equity_curve else 1.0
|
||||
max_dd = 0.0
|
||||
for eq in equity_curve:
|
||||
if eq > peak:
|
||||
peak = eq
|
||||
dd = (peak - eq) / peak
|
||||
if dd > max_dd:
|
||||
max_dd = dd
|
||||
print(f"=== {name} ===")
|
||||
print(f"Bars: {bars}")
|
||||
print(f"Trades: {n} (W{wins} / L{losses})")
|
||||
print(f"Strategy return: {strat_return * 100:+.2f}%")
|
||||
print(f"Buy & Hold return: {bh_return * 100:+.2f}%")
|
||||
print(f"Excess over BH: {(strat_return - bh_return) * 100:+.2f}%")
|
||||
print(f"Max drawdown: {max_dd * 100:.2f}%")
|
||||
print(
|
||||
f"Per-trade Sharpe: {sharpe:.2f} "
|
||||
f"(mean {mean_ret:+.4f}, stddev {math.sqrt(var_ret):.4f})"
|
||||
)
|
||||
print(f"Best / worst trade: {best * 100:+.2f}% / {worst * 100:+.2f}%")
|
||||
print()
|
||||
print(
|
||||
"NOTE: Educational example — fees, slippage, funding costs and tax "
|
||||
"effects are simplified or omitted. Past performance is not "
|
||||
"indicative of future results."
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
path = Path(__file__).resolve().parents[1] / "data" / "btcusdt-1h.csv"
|
||||
candles = load_candles(path)
|
||||
|
||||
macd = ta.MACD(12, 26, 9)
|
||||
adx = ta.ADX(14)
|
||||
|
||||
in_position = False
|
||||
entry_price = 0.0
|
||||
closed_trades: list[float] = []
|
||||
equity = 1.0
|
||||
equity_curve: list[float] = []
|
||||
prev_hist_sign: bool | None = None
|
||||
|
||||
for c in candles:
|
||||
macd_out = macd.update(c["close"])
|
||||
adx_out = adx.update(c["high"], c["low"], c["close"])
|
||||
price = c["close"]
|
||||
mtm = equity * (price / entry_price) if in_position else equity
|
||||
equity_curve.append(mtm)
|
||||
|
||||
if macd_out is None or adx_out is None:
|
||||
continue
|
||||
|
||||
# MACD output is a (macd, signal, histogram) tuple/object across
|
||||
# bindings. The Python binding returns a namedtuple.
|
||||
histogram = macd_out[2] if isinstance(macd_out, tuple) else macd_out.histogram
|
||||
adx_value = adx_out[0] if isinstance(adx_out, tuple) else adx_out.adx
|
||||
|
||||
hist_sign = histogram > 0.0
|
||||
cross_up = prev_hist_sign is False and hist_sign
|
||||
cross_down = prev_hist_sign is True and not hist_sign
|
||||
prev_hist_sign = hist_sign
|
||||
|
||||
if not in_position and cross_up and adx_value > ADX_FLOOR:
|
||||
entry_price = price
|
||||
equity *= 1.0 - FEE
|
||||
in_position = True
|
||||
elif in_position and cross_down:
|
||||
trade_ret = price / entry_price - 1.0
|
||||
closed_trades.append(trade_ret)
|
||||
equity *= (1.0 + trade_ret) * (1.0 - FEE)
|
||||
in_position = False
|
||||
|
||||
if in_position:
|
||||
last_price = candles[-1]["close"]
|
||||
trade_ret = last_price / entry_price - 1.0
|
||||
closed_trades.append(trade_ret)
|
||||
equity *= (1.0 + trade_ret) * (1.0 - FEE)
|
||||
|
||||
print_summary(
|
||||
"MACD + ADX Trend Filter (1h, BTCUSDT)",
|
||||
candles[0]["close"],
|
||||
candles[-1]["close"],
|
||||
len(candles),
|
||||
closed_trades,
|
||||
equity,
|
||||
equity_curve,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,148 @@
|
||||
"""Strategy example: RSI mean-reversion on hourly BTCUSDT data.
|
||||
|
||||
Goes long when RSI(14) crosses below 30 (oversold), exits when RSI
|
||||
crosses above 70 (overbought). Position is binary (full-in / full-out),
|
||||
fees are 0.1% per trade (Binance maker tier), no stop-loss.
|
||||
|
||||
Educational example. NOT a recommended trading strategy in real markets.
|
||||
The point is to show how Wickra streaming indicators wire up into a
|
||||
complete signal -> fill -> PnL -> equity loop in a single file.
|
||||
|
||||
Run with::
|
||||
|
||||
python -m examples.python.strategy_rsi_mean_reversion
|
||||
|
||||
Uses the checked-in ``examples/data/btcusdt-1h.csv`` dataset.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import math
|
||||
from pathlib import Path
|
||||
|
||||
import wickra as ta
|
||||
|
||||
FEE = 0.001
|
||||
RSI_PERIOD = 14
|
||||
OVERSOLD = 30.0
|
||||
OVERBOUGHT = 70.0
|
||||
|
||||
|
||||
def load_candles(path: Path) -> list[dict[str, float]]:
|
||||
with path.open() as fh:
|
||||
reader = csv.DictReader(fh)
|
||||
return [
|
||||
{
|
||||
"open": float(r["open"]),
|
||||
"high": float(r["high"]),
|
||||
"low": float(r["low"]),
|
||||
"close": float(r["close"]),
|
||||
"volume": float(r["volume"]),
|
||||
}
|
||||
for r in reader
|
||||
]
|
||||
|
||||
|
||||
def print_summary(
|
||||
name: str,
|
||||
first_price: float,
|
||||
last_price: float,
|
||||
bars: int,
|
||||
closed_trades: list[float],
|
||||
final_equity: float,
|
||||
equity_curve: list[float],
|
||||
) -> None:
|
||||
buy_hold = last_price / first_price
|
||||
strat_return = final_equity - 1.0
|
||||
bh_return = buy_hold - 1.0
|
||||
wins = sum(1 for r in closed_trades if r > 0)
|
||||
losses = sum(1 for r in closed_trades if r < 0)
|
||||
best = max(closed_trades) if closed_trades else 0.0
|
||||
worst = min(closed_trades) if closed_trades else 0.0
|
||||
n = len(closed_trades)
|
||||
mean_ret = sum(closed_trades) / n if n else 0.0
|
||||
var_ret = (
|
||||
sum((r - mean_ret) ** 2 for r in closed_trades) / (n - 1) if n > 1 else 0.0
|
||||
)
|
||||
sharpe = mean_ret / math.sqrt(var_ret) if var_ret > 0 else 0.0
|
||||
peak = equity_curve[0] if equity_curve else 1.0
|
||||
max_dd = 0.0
|
||||
for eq in equity_curve:
|
||||
if eq > peak:
|
||||
peak = eq
|
||||
dd = (peak - eq) / peak
|
||||
if dd > max_dd:
|
||||
max_dd = dd
|
||||
print(f"=== {name} ===")
|
||||
print(f"Bars: {bars}")
|
||||
print(f"Trades: {n} (W{wins} / L{losses})")
|
||||
print(f"Strategy return: {strat_return * 100:+.2f}%")
|
||||
print(f"Buy & Hold return: {bh_return * 100:+.2f}%")
|
||||
print(f"Excess over BH: {(strat_return - bh_return) * 100:+.2f}%")
|
||||
print(f"Max drawdown: {max_dd * 100:.2f}%")
|
||||
print(
|
||||
f"Per-trade Sharpe: {sharpe:.2f} "
|
||||
f"(mean {mean_ret:+.4f}, stddev {math.sqrt(var_ret):.4f})"
|
||||
)
|
||||
print(f"Best / worst trade: {best * 100:+.2f}% / {worst * 100:+.2f}%")
|
||||
print()
|
||||
print(
|
||||
"NOTE: Educational example — fees, slippage, funding costs and tax "
|
||||
"effects are simplified or omitted. Past performance is not "
|
||||
"indicative of future results."
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
path = Path(__file__).resolve().parents[1] / "data" / "btcusdt-1h.csv"
|
||||
candles = load_candles(path)
|
||||
if len(candles) < RSI_PERIOD * 4:
|
||||
raise SystemExit(f"dataset too small: {len(candles)}")
|
||||
|
||||
rsi = ta.RSI(RSI_PERIOD)
|
||||
|
||||
in_position = False
|
||||
entry_price = 0.0
|
||||
closed_trades: list[float] = []
|
||||
equity = 1.0
|
||||
equity_curve: list[float] = []
|
||||
|
||||
for c in candles:
|
||||
rsi_val = rsi.update(c["close"])
|
||||
price = c["close"]
|
||||
mtm = equity * (price / entry_price) if in_position else equity
|
||||
equity_curve.append(mtm)
|
||||
|
||||
if rsi_val is None:
|
||||
continue
|
||||
|
||||
if not in_position and rsi_val < OVERSOLD:
|
||||
entry_price = price
|
||||
equity *= 1.0 - FEE
|
||||
in_position = True
|
||||
elif in_position and rsi_val > OVERBOUGHT:
|
||||
trade_ret = price / entry_price - 1.0
|
||||
closed_trades.append(trade_ret)
|
||||
equity *= (1.0 + trade_ret) * (1.0 - FEE)
|
||||
in_position = False
|
||||
|
||||
if in_position:
|
||||
last_price = candles[-1]["close"]
|
||||
trade_ret = last_price / entry_price - 1.0
|
||||
closed_trades.append(trade_ret)
|
||||
equity *= (1.0 + trade_ret) * (1.0 - FEE)
|
||||
|
||||
print_summary(
|
||||
"RSI Mean-Reversion (1h, BTCUSDT)",
|
||||
candles[0]["close"],
|
||||
candles[-1]["close"],
|
||||
len(candles),
|
||||
closed_trades,
|
||||
equity,
|
||||
equity_curve,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user