5.3 KiB
5.3 KiB
In [ ]:
import ferro_ta.config as config
import numpy as np
from ferro_ta.backtest import backtest
from ferro_ta.pipeline import Pipeline
from ferro_ta import BBANDS, EMA, RSI, SMA
# Synthetic data
np.random.seed(42)
n = 300
close = np.cumprod(1 + np.random.randn(n) * 0.01) * 100
volume = np.random.randint(1000, 10000, n).astype(float)
print(f"Generated {n} bars, final price: {close[-1]:.2f}")In [ ]:
result = backtest(close, strategy="rsi_30_70", timeperiod=14)
print("Strategy: RSI 30/70")
print(f"Final equity: {result.final_equity:.4f}")
print(f"Number of trades: {result.n_trades}")
print(f"Return: {(result.final_equity - 1.0) * 100:.2f}%")In [ ]:
result2 = backtest(close, strategy="sma_crossover", fast=10, slow=30)
print("Strategy: SMA Crossover (10/30)")
print(f"Final equity: {result2.final_equity:.4f}")
print(f"Number of trades: {result2.n_trades}")In [ ]:
# Set global defaults
config.set_default("timeperiod", 20) # global default for all indicators
config.set_default("RSI.timeperiod", 14) # RSI-specific override
print("Current defaults:", config.list_defaults())
print("RSI defaults:", config.get_defaults_for("RSI"))
print("SMA defaults:", config.get_defaults_for("SMA"))In [ ]:
# Context manager for temporary overrides
with config.Config(timeperiod=5):
temp_default = config.get_default("timeperiod")
print(f"Inside context: timeperiod={temp_default}")
print(f"After context: timeperiod={config.get_default('timeperiod')}") # back to 20
# Clean up
config.reset()In [ ]:
pipe = (
Pipeline()
.add("sma_10", SMA, timeperiod=10)
.add("sma_30", SMA, timeperiod=30)
.add("ema_10", EMA, timeperiod=10)
.add("rsi_14", RSI, timeperiod=14)
.add(
"bb",
BBANDS,
output_keys=["bb_upper", "bb_mid", "bb_lower"],
timeperiod=20,
nbdevup=2.0,
nbdevdn=2.0,
)
)
features = pipe.run(close)
print("Feature columns:", list(features.keys()))
# Build a simple feature matrix (last 5 complete rows)
valid_start = 30 # warmup
feature_matrix = np.column_stack([v[valid_start:] for v in features.values()])
print(f"Feature matrix shape: {feature_matrix.shape}")In [ ]:
# Signal: long when RSI < 40 AND close > SMA_30; flat otherwise
rsi_vals = features["rsi_14"]
sma30_vals = features["sma_30"]
signal = np.where((rsi_vals < 40) & (close > sma30_vals), 1.0, 0.0)
position = np.roll(signal, 1) # trade on next bar open
position[0] = 0.0
returns = np.diff(close) / close[:-1]
strategy_returns = returns * position[1:]
equity = np.cumprod(1 + strategy_returns)
print(f"Final equity: {equity[-1]:.4f}")
print(f"Number of signal bars: {int(signal.sum())}")