gavindiaz/forex-quant-dashboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: forex quant dashboard with momentum strategy + TA-Lib free impl

Browse Source
This commit is contained in:
addychai355-create
2026-05-12 21:42:26 +08:00
commit 24d2f4feed
15 changed files with 1643 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+7
View File
@@ -0,0 +1,7 @@
# OANDA API (free demo: https://www.oanda.com/demo-account/)
OANDA_API_KEY=your_api_key_here
OANDA_ACCOUNT_ID=your_account_id_here
OANDA_ENVIRONMENT=practice # practice or live
# Optional: Alpha Vantage for free FX data (limited)
ALPHA_VANTAGE_KEY=your_key_here
.gitignore
+15
View File
@@ -0,0 +1,15 @@
# Forex Quant Trading Project
.env
__pycache__/
*.pyc
.venv/
*.egg-info/
dist/
build/
.vscode/
notebooks/.ipynb_checkpoints/
data/raw/*
data/processed/*
!data/raw/.gitkeep
!data/processed/.gitkeep
logs/*.log
backtests/.ipynb_checkpoints/Untitled-checkpoint.ipynb
+6
View File
@@ -0,0 +1,6 @@
{
"cells": [],
"metadata": {},
"nbformat": 4,
"nbformat_minor": 5
}
backtests/.ipynb_checkpoints/runner-checkpoint.py
+179
View File
@@ -0,0 +1,179 @@
"""
Backtest Runner — VectorBT based backtesting
Usage:
python backtests/runner.py --pair EUR_USD --tf 1h --years 2
python backtests/runner.py --pair GBP_USD --tf 1h --years 3 --plot
python backtests/runner.py --multi EUR_USD GBP_USD USD_JPY
"""
import argparse
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
import numpy as np
import pandas as pd
from data.fx_data import get_forex_data, AVAILABLE_PAIRS
from strategies.momentum import add_indicators, generate_signals
# Silence TensorFlow warnings if any
import warnings
warnings.filterwarnings("ignore")
def run_backtest(
pair: str = "EUR_USD",
tf: str = "1h",
years_back: int = 2,
plot: bool = False,
spread_pips: float = 1.0,
cash: float = 10_000,
) -> dict:
"""Run a backtest on the momentum strategy."""
print(f"\n📥 Loading {pair} @ {tf} ({years_back}yr)...")
df = get_forex_data(pair, tf, years_back=years_back, cache=True)
if df.empty or len(df) < 100:
print(f"❌ Not enough data for {pair} (got {len(df)} rows)")
return {}
df = df.sort_values("time").reset_index(drop=True)
print(f" Candles: {len(df):,} ({df['time'].min():%Y-%m-%d} → {df['time'].max():%Y-%m-%d})")
# Generate signals
df = add_indicators(df)
df = generate_signals(df, use_macd_filter=True)
# Vectorized backtest
close = df["close"].values
position = df["position"].values
returns = np.diff(close) / close[:-1]
strategy_returns = position[:-1] * returns
# Transaction costs
trades = np.abs(np.diff(position))
spread_decimal = spread_pips * 0.0001 if "JPY" not in pair else spread_pips * 0.01
strategy_returns -= trades * spread_decimal / close[:-1]
# Equity curve
equity = cash * np.cumprod(1 + np.concatenate([[0], strategy_returns]))
# Metrics
total_return = (equity[-1] / cash) - 1
buy_hold_return = (close[-1] / close[0]) - 1
sharpe = 0.0
if strategy_returns.std() > 0:
# Annualize based on data frequency
ann_factor = 252 if tf == "1d" else 252 * 24 if "h" in tf else 252 * 24 * 60
sharpe = round((strategy_returns.mean() / strategy_returns.std()) * np.sqrt(ann_factor), 2)
# Max drawdown
peak = np.maximum.accumulate(equity)
dd = (equity - peak) / peak
max_dd = dd.min()
# Win rate
trade_runs = strategy_returns[trades > 0]
win_rate = (trade_runs > 0).mean() if len(trade_runs) > 2 else 0.0
# Count trades: pair of entry (0→1) + exit (1→0)
num_trades = max(int(np.sum(trades) / 2), 1 if int(np.sum(trades)) >= 1 else 0)
exposure = np.mean(position != 0) * 100
print(f"\n📊 Results for {pair} @ {tf}")
print(f" Total Return: {total_return:+.2%}")
print(f" Buy & Hold: {buy_hold_return:+.2%}")
print(f" Sharpe: {sharpe}")
print(f" Max DD: {max_dd:.2%}")
print(f" Win Rate: {win_rate:.1%}")
print(f" Trades: {num_trades}")
print(f" Exposure: {exposure:.1f}%")
if plot:
try:
import matplotlib.pyplot as plt
fig, axes = plt.subplots(3, 1, figsize=(14, 8), sharex=True)
axes[0].plot(df["time"], close, label=f"{pair} Close", alpha=0.7)
axes[0].set_title(f"{pair} @ {tf} — Momentum Strategy")
axes[0].legend()
axes[0].grid(alpha=0.3)
axes[1].plot(df["time"][1:], strategy_returns, label="Strategy Returns", alpha=0.5)
axes[1].axhline(0, color="black", linewidth=0.5)
axes[1].legend()
axes[1].grid(alpha=0.3)
axes[2].plot(equity, label="Equity", color="green")
axes[2].fill_between(range(len(equity)), equity, cash, alpha=0.1, color="green")
axes[2].axhline(cash, color="gray", linestyle="--", alpha=0.5)
axes[2].legend()
axes[2].grid(alpha=0.3)
plt.tight_layout()
plt.show()
except Exception as e:
print(f" Plot error: {e}")
return {
"pair": pair,
"tf": tf,
"total_return": round(total_return * 100, 2),
"buy_hold_return": round(buy_hold_return * 100, 2),
"sharpe": sharpe,
"max_dd": round(max_dd * 100, 2),
"win_rate": round(win_rate * 100, 1),
"trades": num_trades,
"exposure_pct": round(exposure, 1),
}
def run_multi_backtest(pairs: list[str], tf: str = "1h", years: int = 2):
"""Run backtest across multiple pairs and compare."""
all_results = []
for pair in pairs:
print(f"\n{'='*55}")
r = run_backtest(pair, tf, years, plot=False)
if r:
all_results.append(r)
print(f"{'='*55}")
if all_results:
results_df = pd.DataFrame(all_results)
results_df = results_df.sort_values("sharpe", ascending=False)
print(f"\n🏆 Multi-Pair Summary ({tf}, {years}yr)")
print(results_df.to_string(index=False))
out_path = Path("logs") / f"multi_{tf}_{years}yr.csv"
out_path.parent.mkdir(exist_ok=True)
results_df.to_csv(out_path, index=False)
print(f"\nSaved to {out_path}")
return all_results
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Forex Quant Backtest")
parser.add_argument("--pair", default="EUR_USD", help="Forex pair")
parser.add_argument("--tf", default="1h", help="Timeframe: 1m, 5m, 15m, 30m, 1h, 4h, 1d")
parser.add_argument("--years", type=int, default=2, help="Years of history")
parser.add_argument("--plot", action="store_true", help="Show plot")
parser.add_argument("--multi", nargs="*", help="Pairs for multi-run (e.g. EUR_USD GBP_USD)")
parser.add_argument("--spread", type=float, default=1.0, help="Spread in pips")
parser.add_argument("--cash", type=float, default=10_000, help="Starting capital")
parser.add_argument("--list-pairs", action="store_true", help="List available pairs")
args = parser.parse_args()
if args.list_pairs:
print("Available pairs:")
for p in AVAILABLE_PAIRS:
print(f" • {p}")
sys.exit(0)
if args.multi:
run_multi_backtest(args.multi, args.tf, args.years)
else:
run_backtest(args.pair, args.tf, args.years, args.plot, args.spread, args.cash)
backtests/Untitled.ipynb
+6
View File
@@ -0,0 +1,6 @@
{
"cells": [],
"metadata": {},
"nbformat": 4,
"nbformat_minor": 5
}
backtests/runner.py
+179
View File
@@ -0,0 +1,179 @@
"""
Backtest Runner — VectorBT based backtesting
Usage:
python backtests/runner.py --pair EUR_USD --tf 1h --years 2
python backtests/runner.py --pair GBP_USD --tf 1h --years 3 --plot
python backtests/runner.py --multi EUR_USD GBP_USD USD_JPY
"""
import argparse
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
import numpy as np
import pandas as pd
from data.fx_data import get_forex_data, AVAILABLE_PAIRS
from strategies.momentum import add_indicators, generate_signals
# Silence TensorFlow warnings if any
import warnings
warnings.filterwarnings("ignore")
def run_backtest(
pair: str = "EUR_USD",
tf: str = "1h",
years_back: int = 2,
plot: bool = False,
spread_pips: float = 1.0,
cash: float = 10_000,
) -> dict:
"""Run a backtest on the momentum strategy."""
print(f"\n📥 Loading {pair} @ {tf} ({years_back}yr)...")
df = get_forex_data(pair, tf, years_back=years_back, cache=True)
if df.empty or len(df) < 100:
print(f"❌ Not enough data for {pair} (got {len(df)} rows)")
return {}
df = df.sort_values("time").reset_index(drop=True)
print(f" Candles: {len(df):,} ({df['time'].min():%Y-%m-%d} → {df['time'].max():%Y-%m-%d})")
# Generate signals
df = add_indicators(df)
df = generate_signals(df, use_macd_filter=True)
# Vectorized backtest
close = df["close"].values
position = df["position"].values
returns = np.diff(close) / close[:-1]
strategy_returns = position[:-1] * returns
# Transaction costs
trades = np.abs(np.diff(position))
spread_decimal = spread_pips * 0.0001 if "JPY" not in pair else spread_pips * 0.01
strategy_returns -= trades * spread_decimal / close[:-1]
# Equity curve
equity = cash * np.cumprod(1 + np.concatenate([[0], strategy_returns]))
# Metrics
total_return = (equity[-1] / cash) - 1
buy_hold_return = (close[-1] / close[0]) - 1
sharpe = 0.0
if strategy_returns.std() > 0:
# Annualize based on data frequency
ann_factor = 252 if tf == "1d" else 252 * 24 if "h" in tf else 252 * 24 * 60
sharpe = round((strategy_returns.mean() / strategy_returns.std()) * np.sqrt(ann_factor), 2)
# Max drawdown
peak = np.maximum.accumulate(equity)
dd = (equity - peak) / peak
max_dd = dd.min()
# Win rate
trade_runs = strategy_returns[trades > 0]
win_rate = (trade_runs > 0).mean() if len(trade_runs) > 2 else 0.0
# Count trades: pair of entry (0→1) + exit (1→0)
num_trades = max(int(np.sum(trades) / 2), 1 if int(np.sum(trades)) >= 1 else 0)
exposure = np.mean(position != 0) * 100
print(f"\n📊 Results for {pair} @ {tf}")
print(f" Total Return: {total_return:+.2%}")
print(f" Buy & Hold: {buy_hold_return:+.2%}")
print(f" Sharpe: {sharpe}")
print(f" Max DD: {max_dd:.2%}")
print(f" Win Rate: {win_rate:.1%}")
print(f" Trades: {num_trades}")
print(f" Exposure: {exposure:.1f}%")
if plot:
try:
import matplotlib.pyplot as plt
fig, axes = plt.subplots(3, 1, figsize=(14, 8), sharex=True)
axes[0].plot(df["time"], close, label=f"{pair} Close", alpha=0.7)
axes[0].set_title(f"{pair} @ {tf} — Momentum Strategy")
axes[0].legend()
axes[0].grid(alpha=0.3)
axes[1].plot(df["time"][1:], strategy_returns, label="Strategy Returns", alpha=0.5)
axes[1].axhline(0, color="black", linewidth=0.5)
axes[1].legend()
axes[1].grid(alpha=0.3)
axes[2].plot(equity, label="Equity", color="green")
axes[2].fill_between(range(len(equity)), equity, cash, alpha=0.1, color="green")
axes[2].axhline(cash, color="gray", linestyle="--", alpha=0.5)
axes[2].legend()
axes[2].grid(alpha=0.3)
plt.tight_layout()
plt.show()
except Exception as e:
print(f" Plot error: {e}")
return {
"pair": pair,
"tf": tf,
"total_return": round(total_return * 100, 2),
"buy_hold_return": round(buy_hold_return * 100, 2),
"sharpe": sharpe,
"max_dd": round(max_dd * 100, 2),
"win_rate": round(win_rate * 100, 1),
"trades": num_trades,
"exposure_pct": round(exposure, 1),
}
def run_multi_backtest(pairs: list[str], tf: str = "1h", years: int = 2):
"""Run backtest across multiple pairs and compare."""
all_results = []
for pair in pairs:
print(f"\n{'='*55}")
r = run_backtest(pair, tf, years, plot=False)
if r:
all_results.append(r)
print(f"{'='*55}")
if all_results:
results_df = pd.DataFrame(all_results)
results_df = results_df.sort_values("sharpe", ascending=False)
print(f"\n🏆 Multi-Pair Summary ({tf}, {years}yr)")
print(results_df.to_string(index=False))
out_path = Path("logs") / f"multi_{tf}_{years}yr.csv"
out_path.parent.mkdir(exist_ok=True)
results_df.to_csv(out_path, index=False)
print(f"\nSaved to {out_path}")
return all_results
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Forex Quant Backtest")
parser.add_argument("--pair", default="EUR_USD", help="Forex pair")
parser.add_argument("--tf", default="1h", help="Timeframe: 1m, 5m, 15m, 30m, 1h, 4h, 1d")
parser.add_argument("--years", type=int, default=2, help="Years of history")
parser.add_argument("--plot", action="store_true", help="Show plot")
parser.add_argument("--multi", nargs="*", help="Pairs for multi-run (e.g. EUR_USD GBP_USD)")
parser.add_argument("--spread", type=float, default=1.0, help="Spread in pips")
parser.add_argument("--cash", type=float, default=10_000, help="Starting capital")
parser.add_argument("--list-pairs", action="store_true", help="List available pairs")
args = parser.parse_args()
if args.list_pairs:
print("Available pairs:")
for p in AVAILABLE_PAIRS:
print(f" • {p}")
sys.exit(0)
if args.multi:
run_multi_backtest(args.multi, args.tf, args.years)
else:
run_backtest(args.pair, args.tf, args.years, args.plot, args.spread, args.cash)
config.py
+56
View File
@@ -0,0 +1,56 @@
"""
Forex Quant Trading — Configuration
"""
import os
from pathlib import Path
ROOT = Path(__file__).parent
DATA_DIR = ROOT / "data"
RAW_DIR = DATA_DIR / "raw"
PROCESSED_DIR = DATA_DIR / "processed"
LOGS_DIR = ROOT / "logs"
# Ensure directories exist
for d in [RAW_DIR, PROCESSED_DIR, LOGS_DIR]:
d.mkdir(parents=True, exist_ok=True)
# === Trading Parameters ===
TIMEFRAMES = {
"1m": "M1",
"5m": "M5",
"15m": "M15",
"30m": "M30",
"1h": "H1",
"4h": "H4",
"1d": "D1",
}
# Major forex pairs to watch
MAJOR_PAIRS = [
"EUR_USD", "GBP_USD", "USD_JPY", "USD_CHF",
"AUD_USD", "USD_CAD", "NZD_USD",
]
# Cross pairs
CROSS_PAIRS = [
"EUR_GBP", "EUR_JPY", "GBP_JPY", "EUR_AUD",
"AUD_JPY", "CHF_JPY", "EUR_CHF",
]
DEFAULT_PAIR = "EUR_USD"
DEFAULT_TF = "H1"
# === Risk Management ===
RISK_PER_TRADE = 0.01 # 1% risk per trade
MAX_SPREAD_PIPS = 2.0 # Max acceptable spread
STOP_LOSS_ATR_MULT = 1.5 # SL = ATR * this
TAKE_PROFIT_ATR_MULT = 3.0 # TP = ATR * this
# === Data Sources ===
# OANDA practice account — sign up free: https://www.oanda.com/demo-account/
OANDA_KEY = os.getenv("OANDA_API_KEY", "")
OANDA_ACCOUNT = os.getenv("OANDA_ACCOUNT_ID", "")
OANDA_ENV = os.getenv("OANDA_ENVIRONMENT", "practice")
# === Mode ===
DRY_RUN = True # paper trade until you're confident
dashboard/README.md
+58
View File
@@ -0,0 +1,58 @@
# Forex Quant Dashboard
Real-time monitoring dashboard for your forex quant trading strategy.
Free to run locally or deploy to Streamlit Community Cloud.
## Quick Start (Local)
```bash
cd ~/forex-quant
source .venv/bin/activate
streamlit run dashboard/app.py
```
Open http://localhost:8501 in your browser.
## Deploy for Free (Remote Access)
### Option 1: Streamlit Community Cloud ⭐ (recommended)
1. Push `dashboard/` to a GitHub repo
2. Go to https://streamlit.io/cloud
3. Sign in with GitHub
4. Click "New app" → select your repo
5. Main file path: `dashboard/app.py`
6. Deploy → your app is live at `https://your-app.streamlit.app`
**No credit card needed. Free forever.**
- 1 app included
- Unlimited team members
- Community support
### Option 2: Local + Cloudflare Tunnel
```bash
# Install cloudflared
brew install cloudflare/cloudflared/cloudflared
# Run dashboard
streamlit run dashboard/app.py --server.port 8501
# In another terminal, expose it
cloudflared tunnel --url http://localhost:8501
```
You get a `*.trycloudflare.com` URL — accessible from anywhere.
## Dashboard Features
- **Live prices** — all major forex pairs
- **Strategy signals** — MACD-confirmed momentum
- **Performance metrics** — Sharpe, drawdown, win rate
- **Equity curve** — strategy vs buy & hold
- **Multi-pair comparison** — find what's working
- **Adjustable parameters** — tweak in real-time
## Data Source
All data from Yahoo Finance — **free, no API key required**.
dashboard/app.py
+419
View File
@@ -0,0 +1,419 @@
"""
Forex Quant Dashboard — Streamlit App
Monitor signals, performance, and live prices from anywhere.
Deploy to Streamlit Community Cloud for free:
1. Push this folder to GitHub
2. Go to https://streamlit.io/cloud
3. Connect repo → Deploy
"""
import sys
from pathlib import Path
# Add project root to path
sys.path.insert(0, str(Path(__file__).parent.parent))
import warnings
warnings.filterwarnings("ignore")
import streamlit as st
import pandas as pd
import numpy as np
import plotly.graph_objects as go
import plotly.express as px
from plotly.subplots import make_subplots
from datetime import datetime, timedelta, timezone
from data.fx_data import get_forex_data, AVAILABLE_PAIRS
from strategies.momentum import add_indicators, generate_signals, calculate_performance
st.set_page_config(
page_title="Forex Quant Monitor",
page_icon="📊",
layout="wide",
initial_sidebar_state="expanded",
)
# ─── Color scheme ───
COLORS = {
"bg": "#0E1117",
"card": "#1A1D23",
"green": "#00C853",
"red": "#FF1744",
"blue": "#448AFF",
"yellow": "#FFD600",
"text": "#E0E0E0",
}
st.markdown("""
<style>
.stApp { background-color: #0E1117; }
.css-1r6slb0 { background-color: #1A1D23; }
.metric-card {
background: #1A1D23;
padding: 1rem;
border-radius: 8px;
border: 1px solid #2D3039;
}
.metric-value { font-size: 1.8rem; font-weight: 700; }
.metric-label { font-size: 0.8rem; color: #9E9E9E; }
.positive { color: #00C853; }
.negative { color: #FF1744; }
</style>
""", unsafe_allow_html=True)
# ─── Sidebar ───
st.sidebar.title("📊 Forex Monitor")
st.sidebar.markdown("---")
# Pair selector
pair = st.sidebar.selectbox("Pair", AVAILABLE_PAIRS, index=0)
# Timeframe
tf_options = {"1m": "1 Min", "5m": "5 Min", "15m": "15 Min", "30m": "30 Min",
"1h": "1 Hour", "4h": "4 Hour", "1d": "1 Day"}
tf = st.sidebar.selectbox("Timeframe", list(tf_options.keys()),
format_func=lambda x: tf_options[x], index=4)
# Date range
years_back = st.sidebar.slider("History", 1, 5, 2)
st.sidebar.markdown("---")
st.sidebar.subheader("Strategy Params")
atr_min = st.sidebar.slider("Min ATR %", 0.01, 0.50, 0.05, 0.01)
use_macd = st.sidebar.checkbox("MACD Filter", value=True)
st.sidebar.markdown("---")
st.sidebar.caption("Data: Yahoo Finance (free)")
st.sidebar.caption(f"Updated: {datetime.now(timezone.utc):%Y-%m-%d %H:%M} UTC")
auto_refresh = st.sidebar.checkbox("Auto-refresh every 60s", value=False)
if auto_refresh:
st.sidebar.info("🔄 Refreshing...")
st.rerun(60)
# ─── Load Data ───
@st.cache_data(ttl=300) # 5 min cache
def load_data(pr, tf_str, yrs):
"""Load forex data with caching."""
df = get_forex_data(pr, tf_str, years_back=yrs, cache=True, source="yahoo")
if df.empty or len(df) < 50:
return None
df = add_indicators(df)
df = generate_signals(df, atr_min_pct=atr_min, use_macd_filter=use_macd)
return df
@st.cache_data(ttl=300)
def load_all_pairs_data(tf_str, yrs):
"""Load latest data for all pairs (for overview)."""
results = {}
for p in AVAILABLE_PAIRS:
try:
df = get_forex_data(p, tf_str, years_back=yrs, cache=True, source="yahoo")
if not df.empty and len(df) > 20:
results[p] = df
except Exception:
continue
return results
# ─── Main Dashboard ───
# Row 1: Live Prices Overview
st.subheader("💰 Live Prices Overview")
with st.spinner("Loading market data..."):
all_data = load_all_pairs_data("1h", 1)
if all_data:
cols = st.columns(4)
for i, (p, df) in enumerate(sorted(all_data.items())):
latest = df.iloc[-1]
prev = df.iloc[-2]
change = latest["close"] - prev["close"]
change_pct = change / prev["close"] * 100
with cols[i % 4]:
color = COLORS["green"] if change >= 0 else COLORS["red"]
arrow = "▲" if change >= 0 else "▼"
st.markdown(f"""
<div class="metric-card">
<div class="metric-label">{p.replace('_', '/')}</div>
<div class="metric-value">{latest['close']:.5f}</div>
<div style="color:{color}">
{arrow} {change:.5f} ({change_pct:+.3f}%)
</div>
</div>
""", unsafe_allow_html=True)
else:
st.warning("Could not load price data. Check internet connection.")
st.markdown("---")
# Row 2: Main Strategy Chart
st.subheader(f"📈 {pair.replace('_', '/')} — Strategy Analysis")
data = load_data(pair, tf, years_back)
if data is not None:
col1, col2 = st.columns([2, 1])
with col1:
# Price + signals chart
fig = make_subplots(
rows=3, cols=1,
shared_xaxes=True,
vertical_spacing=0.05,
row_heights=[0.55, 0.25, 0.20],
subplot_titles=(f"{pair.replace('_', '/')} Price & Signals", "MACD", "RSI"),
)
# Candlestick chart
fig.add_trace(go.Candlestick(
x=data["time"],
open=data["open"],
high=data["high"],
low=data["low"],
close=data["close"],
name="Price",
showlegend=False,
), row=1, col=1)
# Buy/Sell markers
buy_signals = data[data["signal"] == 1]
fig.add_trace(go.Scatter(
x=buy_signals["time"],
y=buy_signals["close"],
mode="markers",
marker=dict(symbol="triangle-up", size=12, color=COLORS["green"]),
name="Enter Long",
), row=1, col=1)
# MAs
fig.add_trace(go.Scatter(
x=data["time"], y=data["ma_fast"],
line=dict(color=COLORS["blue"], width=1),
name="MA-8",
), row=1, col=1)
fig.add_trace(go.Scatter(
x=data["time"], y=data["ma_mid"],
line=dict(color=COLORS["yellow"], width=1),
name="MA-21",
), row=1, col=1)
# MACD
fig.add_trace(go.Bar(
x=data["time"], y=data["macd_hist"],
marker_color=np.where(data["macd_hist"] >= 0, COLORS["green"], COLORS["red"]),
name="MACD Hist",
), row=2, col=1)
fig.add_trace(go.Scatter(
x=data["time"], y=data["macd"],
line=dict(color=COLORS["blue"], width=1.5),
name="MACD",
), row=2, col=1)
fig.add_trace(go.Scatter(
x=data["time"], y=data["macd_signal"],
line=dict(color=COLORS["yellow"], width=1.5),
name="Signal",
), row=2, col=1)
# RSI
fig.add_trace(go.Scatter(
x=data["time"], y=data["rsi"],
line=dict(color=COLORS["blue"], width=1.5),
name="RSI",
), row=3, col=1)
fig.add_hline(y=70, line_dash="dash", line_color=COLORS["red"], row=3, col=1)
fig.add_hline(y=30, line_dash="dash", line_color=COLORS["green"], row=3, col=1)
fig.update_layout(
height=650,
template="plotly_dark",
hovermode="x unified",
margin=dict(l=0, r=0, t=30, b=0),
legend=dict(orientation="h", y=1.02, x=0),
)
fig.update_xaxes(rangeslider_visible=False)
st.plotly_chart(fig, use_container_width=True)
with col2:
# Strategy metrics
perf = calculate_performance(data)
st.markdown("### 📊 Performance")
metrics = [
("Return", f"{perf['total_return_pct']:+.2f}%", "positive" if perf['total_return_pct'] > 0 else "negative"),
("Buy & Hold", f"{perf['buy_hold_return_pct']:+.2f}%", "positive" if perf['buy_hold_return_pct'] > 0 else "negative"),
("Sharpe", f"{perf['sharpe_ratio']}", "positive" if perf['sharpe_ratio'] > 1 else "neutral" if perf['sharpe_ratio'] > 0 else "negative"),
("Max Drawdown", f"{perf['max_drawdown_pct']:.2f}%", "negative"),
("Win Rate", f"{perf['win_rate_pct']:.1f}%", "positive" if perf['win_rate_pct'] > 50 else "negative"),
("Trades", f"{perf['num_trades']}", "neutral"),
("Exposure", f"{perf['exposure_pct']:.1f}%", "neutral"),
]
for label, value, cls in metrics:
st.markdown(f"""
<div style="display:flex; justify-content:space-between; padding:4px 0; border-bottom:1px solid #2D3039;">
<span style="color:#9E9E9E;">{label}</span>
<span class="{cls}" style="font-weight:600;">{value}</span>
</div>
""", unsafe_allow_html=True)
st.markdown("---")
# Current signal
latest_signal = data["signal"].iloc[-1]
latest_position = data["position"].iloc[-1]
latest_rsi = data["rsi"].iloc[-1]
latest_atr = data["atr_pct"].iloc[-1]
st.markdown("### 🔔 Current Status")
signal_icon = "🟢" if latest_position == 1 else "🔴" if latest_position == -1 else "⚪"
signal_text = "LONG" if latest_position == 1 else "SHORT" if latest_position == -1 else "FLAT"
st.markdown(f"""
<div class="metric-card" style="text-align:center;">
<div style="font-size:2rem;">{signal_icon}</div>
<div style="font-size:1.5rem; font-weight:700;">{signal_text}</div>
<div style="color:#9E9E9E;">RSI: {latest_rsi:.1f} | ATR%: {latest_atr:.3f}%</div>
</div>
""", unsafe_allow_html=True)
else:
st.error(f"Could not load data for {pair}. Try a different pair or timeframe.")
st.markdown("---")
# Row 3: Equity Curve + Drawdown
st.subheader("💰 Equity Curve")
if data is not None:
col1, col2 = st.columns([2, 1])
with col1:
# Compute equity curve from signals
df = data.copy()
df["returns"] = df["close"].pct_change()
df["strategy_returns"] = df["position"].shift(1) * df["returns"]
df["trades"] = df["position"].diff().abs().clip(0)
df["strategy_returns"] -= df["trades"] * 0.0001 / df["close"]
df["equity"] = 10000 * (1 + df["strategy_returns"]).cumprod()
df["buy_hold"] = 10000 * (1 + df["returns"]).cumprod()
fig = make_subplots(
rows=2, cols=1,
shared_xaxes=True,
vertical_spacing=0.05,
row_heights=[0.7, 0.3],
)
fig.add_trace(go.Scatter(
x=df["time"], y=df["equity"],
line=dict(color=COLORS["green"], width=2),
name="Strategy",
), row=1, col=1)
fig.add_trace(go.Scatter(
x=df["time"], y=df["buy_hold"],
line=dict(color="#9E9E9E", width=1, dash="dash"),
name="Buy & Hold",
), row=1, col=1)
# Drawdown
peak = df["equity"].expanding().max()
dd = (df["equity"] - peak) / peak * 100
fig.add_trace(go.Scatter(
x=df["time"], y=dd,
fill="tozeroy",
line=dict(color=COLORS["red"], width=1),
name="Drawdown %",
), row=2, col=1)
fig.update_layout(
height=400,
template="plotly_dark",
hovermode="x unified",
margin=dict(l=0, r=0, t=10, b=0),
legend=dict(orientation="h", y=1.02, x=0),
)
st.plotly_chart(fig, use_container_width=True)
with col2:
st.markdown("### 📋 Recent Signals")
sig_cols = ["time", "close", "rsi", "atr_pct", "position", "signal"]
recent = data[sig_cols].tail(20).copy()
recent["position"] = recent["position"].map({1: "LONG", 0: "FLAT", -1: "SHORT"})
recent["signal"] = recent["signal"].map({1: "🟢 BUY", 0: "⚪", -1: "🔴 SELL"})
recent = recent.rename(columns={
"time": "Time", "close": "Price", "rsi": "RSI",
"atr_pct": "ATR%", "position": "Pos", "signal": "Signal"
})
recent["Time"] = recent["Time"].dt.strftime("%m/%d %H:%M")
st.dataframe(recent, use_container_width=True, hide_index=True)
st.markdown("---")
# Row 4: Multi-Pair Heatmap
st.subheader("🌍 Multi-Pair Comparison")
with st.spinner("Loading all pairs..."):
comparison_data = {}
for p in AVAILABLE_PAIRS:
try:
d = load_data(p, "1d", 2)
if d is not None:
perf = calculate_performance(d)
comparison_data[p] = perf
except Exception:
continue
if comparison_data:
comp_df = pd.DataFrame(comparison_data).T
comp_df.index.name = "Pair"
col1, col2 = st.columns([1, 2])
with col1:
metrics_select = st.selectbox("Metric", ["total_return_pct", "sharpe_ratio", "max_drawdown_pct", "win_rate_pct"])
metric_labels = {
"total_return_pct": "Total Return %",
"sharpe_ratio": "Sharpe Ratio",
"max_drawdown_pct": "Max Drawdown %",
"win_rate_pct": "Win Rate %",
}
fig = px.bar(
comp_df.sort_values(metrics_select, ascending=False),
y=metrics_select,
color=metrics_select,
color_continuous_scale=["red", "yellow", "green"],
title=f"{metric_labels[metrics_select]} by Pair",
text_auto=".1f",
)
fig.update_layout(
template="plotly_dark",
height=400,
margin=dict(l=0, r=0, t=30, b=0),
showlegend=False,
)
st.plotly_chart(fig, use_container_width=True)
with col2:
st.markdown("### 📊 Comparison Table")
display = comp_df[[
"total_return_pct", "buy_hold_return_pct",
"sharpe_ratio", "max_drawdown_pct",
"win_rate_pct", "num_trades", "exposure_pct"
]].round(2)
display.columns = [
"Return%", "BH Return%", "Sharpe", "Max DD%",
"Win Rate%", "Trades", "Exposure%"
]
st.dataframe(display, use_container_width=True)
st.markdown("---")
# Footer
st.caption("""
**Forex Quant Monitor** — Data from Yahoo Finance | Strategy: Momentum + Volatility Filter
Built with Streamlit | Deploy free on streamlit.io/cloud
""")
dashboard/packages.txt
+2
View File
@@ -0,0 +1,2 @@
# System dependencies for TA-Lib on Streamlit Cloud
ta-lib
dashboard/requirements.txt
+10
View File
@@ -0,0 +1,10 @@
# Forex Quant Dashboard — Deployment Requirements
# TA-Lib is optional (pandas fallback handles everything)
streamlit>=1.28
pandas>=2.0
numpy>=1.24
plotly>=5.15
yfinance>=0.2.28
scikit-learn>=1.3
scipy>=1.11
python-dotenv>=1.0
data/fx_data.py
+404
View File
@@ -0,0 +1,404 @@
"""
Forex Data Pipeline
====================
Sources (in order of preference):
1. yfinance — free, no API key, works out of the box
2. Dukascopy — free historical tick data (BI5 format)
3. OANDA API — live/practice account, needs API key
"""
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Optional
import numpy as np
import pandas as pd
sys.path.insert(0, str(Path(__file__).parent.parent))
from config import RAW_DIR, OANDA_KEY, OANDA_ACCOUNT, OANDA_ENV
# ──────────────────────────────────────────────
# YAHOO FINANCE — Simplest, most reliable free source
# ──────────────────────────────────────────────
# Yahoo ticker format for forex: EURUSD=X
YAHOO_PAIRS = {
"EUR_USD": "EURUSD=X",
"GBP_USD": "GBPUSD=X",
"USD_JPY": "USDJPY=X",
"USD_CHF": "USDCHF=X",
"AUD_USD": "AUDUSD=X",
"USD_CAD": "USDCAD=X",
"NZD_USD": "NZDUSD=X",
"EUR_GBP": "EURGBP=X",
"EUR_JPY": "EURJPY=X",
"GBP_JPY": "GBPJPY=X",
"EUR_AUD": "EURAUD=X",
"AUD_JPY": "AUDJPY=X",
"CHF_JPY": "CHFJPY=X",
"EUR_CHF": "EURCHF=X",
}
TIMEFRAMES_YAHOO = {
"1m": "1m", "5m": "5m", "15m": "15m", "30m": "30m",
"1h": "60m", "4h": "60m", "1d": "1d",
}
def get_yahoo_data(
pair: str = "EUR_USD",
tf: str = "1h",
years_back: int = 2,
cache: bool = True,
) -> pd.DataFrame:
"""
Download forex data from Yahoo Finance.
Uses yfinance library — no API key needed.
Timeframes: 1m, 5m, 15m, 30m, 1h, 4h, 1d
Max history varies by timeframe (1m = 7 days, 1h = 730 days, 1d = max)
"""
import yfinance as yf
ticker = YAHOO_PAIRS.get(pair, pair.replace("_", "") + "=X")
yahoo_tf = TIMEFRAMES_YAHOO.get(tf, tf)
cache_file = RAW_DIR / f"yahoo_{ticker}_{tf}.parquet"
# Try cache first (handle missing pyarrow gracefully)
if cache:
try:
if cache_file.exists():
df = pd.read_parquet(cache_file)
last_dt = df["time"].max()
if last_dt and pd.Timestamp.now(timezone.utc) - last_dt < timedelta(hours=2):
return df
except Exception:
pass # cache read failed, re-download
try:
# Determine period and interval based on timeframe
if tf in ("1m", "5m"):
period = "7d" if tf == "1m" else "1mo"
elif tf in ("15m", "30m"):
period = "3mo"
elif tf in ("1h",):
period = "2y" # Max for 60m
elif tf in ("4h",):
period = "max" # Max for 60m too
tf = "1h"
yahoo_tf = "60m"
else:
period = f"{years_back}y"
data = yf.download(
tickers=ticker,
period=period,
interval=yahoo_tf,
progress=False,
auto_adjust=True,
)
if data.empty:
print(f" ⚠️ No data from Yahoo for {ticker}")
return pd.DataFrame()
# Flatten multi-level columns if needed
if isinstance(data.columns, pd.MultiIndex):
data.columns = data.columns.get_level_values(0)
df = data.reset_index()
df.columns = [c.lower().strip() for c in df.columns]
# Rename columns
col_map = {
"datetime": "time", "datetime": "time",
"open": "open", "high": "high", "low": "low",
"close": "close", "volume": "volume",
}
df = df.rename(columns={k: v for k, v in col_map.items() if k in df.columns})
df["pair"] = pair
# Ensure time column is datetime
if "time" not in df.columns and "index" in df.columns:
df = df.rename(columns={"index": "time"})
if "time" not in df.columns and len(df.columns) > 0:
# Try first column
first_col = df.columns[0]
df = df.rename(columns={first_col: "time"})
df["time"] = pd.to_datetime(df["time"])
df = df.sort_values("time").reset_index(drop=True)
if cache:
try:
cache_file.parent.mkdir(parents=True, exist_ok=True)
df.to_parquet(cache_file, index=False)
except Exception:
pass
return df
except Exception as e:
print(f" ⚠️ Yahoo Finance error for {ticker}: {e}")
return pd.DataFrame()
# ──────────────────────────────────────────────
# OANDA API — Live/Recent Data (needs API key)
# ──────────────────────────────────────────────
OANDA_BASE = {
"practice": "https://api-fxpractice.oanda.com",
"live": "https://api-fxtrade.oanda.com",
}
def _oanda_headers() -> dict:
return {
"Authorization": f"Bearer {OANDA_KEY}",
"Content-Type": "application/json",
}
def get_oanda_candles(
pair: str = "EUR_USD",
tf: str = "H1",
count: int = 500,
) -> pd.DataFrame:
"""Fetch recent candles from OANDA REST API."""
import requests
if not OANDA_KEY:
raise ValueError("OANDA_API_KEY not set. Add it to .env or use yfinance/Dukascopy.")
base = OANDA_BASE.get(OANDA_ENV, OANDA_BASE["practice"])
url = f"{base}/v3/instruments/{pair}/candles"
params = {
"granularity": tf,
"count": min(count, 5000),
"price": "M",
}
resp = requests.get(url, headers=_oanda_headers(), params=params, timeout=15)
resp.raise_for_status()
data = resp.json()
rows = []
for c in data.get("candles", []):
rows.append({
"time": c["time"].replace("Z", "+00:00"),
"open": float(c["mid"]["o"]),
"high": float(c["mid"]["h"]),
"low": float(c["mid"]["l"]),
"close": float(c["mid"]["c"]),
"volume": int(c["volume"]),
"pair": pair,
})
df = pd.DataFrame(rows)
if not df.empty:
df["time"] = pd.to_datetime(df["time"])
return df
def get_oanda_price(pair: str = "EUR_USD") -> dict:
"""Get current live price from OANDA."""
import requests
base = OANDA_BASE.get(OANDA_ENV, OANDA_BASE["practice"])
url = f"{base}/v3/accounts/{OANDA_ACCOUNT}/pricing"
params = {"instruments": pair}
resp = requests.get(url, headers=_oanda_headers(), params=params, timeout=10)
resp.raise_for_status()
return resp.json()
# ──────────────────────────────────────────────
# DUKASCOPY — Free Historical Data (no API key)
# ──────────────────────────────────────────────
DUKASCOPY_SYMBOLS = {k.replace("_", ""): v for k, v in YAHOO_PAIRS.items()}
DUKASCOPY_SYMBOLS = {v.replace("=X", ""): v.replace("=X", "")
for v in YAHOO_PAIRS.values()}
# Build proper mapping
DUKASCOPY_SYMBOLS = {}
for our, yahoo in YAHOO_PAIRS.items():
sym = yahoo.replace("=X", "")
DUKASCOPY_SYMBOLS[our] = sym
TF_MAP = {
"M1": 60, "M5": 300, "M15": 900, "M30": 1800,
"H1": 3600, "H4": 14400, "D1": 86400,
}
def get_dukascopy_month(
pair: str, year: int, month: int, cache: bool = True
) -> pd.DataFrame:
"""
Download one month of 1-minute OHLC from Dukascopy.
Returns DataFrame with time, open, high, low, close, volume.
NOTE: Dukascopy may reject some requests — yfinance is more reliable.
"""
import certifi
import ssl
import urllib.request, urllib.error
sym = DUKASCOPY_SYMBOLS.get(pair, pair.replace("_", ""))
cache_file = RAW_DIR / f"duka_{sym}_{year}_{month:02d}.parquet"
if cache and cache_file.exists():
return pd.read_parquet(cache_file)
url = (
f"https://data.dukascopy.com/datafeed/{sym}/"
f"{year:04d}/{month-1:02d}/"
f"{year}{month:02d}.zip"
)
ctx = ssl.create_default_context(cafile=certifi.where())
try:
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=30, context=ctx) as resp:
data = resp.read()
except (urllib.error.HTTPError, urllib.error.URLError) as e:
print(f" ⚠️ No Dukascopy data for {sym} {year}-{month:02d}: {e}")
return pd.DataFrame()
# Parse BI5 binary format
import gzip, struct
try:
decompressed = gzip.decompress(data)
except Exception:
decompressed = data
n_records = len(decompressed) // 20
if n_records == 0:
return pd.DataFrame()
fmt = ">" + "i" * (n_records * 5)
vals = struct.unpack(fmt, decompressed[: n_records * 20])
arr = np.array(vals, dtype=np.int64).reshape(-1, 5)
df = pd.DataFrame(arr, columns=["time_offset", "open", "close", "high", "low", "volume"])
month_start = datetime(year, month, 1, tzinfo=timezone.utc)
df["time"] = month_start + pd.to_timedelta(df["time_offset"], unit="s")
is_jpy = sym.endswith("JPY")
scale = 100.0 if is_jpy else 10000.0
for col in ["open", "high", "low", "close"]:
df[col] = df[col].astype(float) / scale
df["pair"] = pair
df = df[["time", "pair", "open", "high", "low", "close", "volume"]].sort_values("time")
if cache:
cache_file.parent.mkdir(parents=True, exist_ok=True)
df.to_parquet(cache_file, index=False)
return df
# ──────────────────────────────────────────────
# Unified interface
# ──────────────────────────────────────────────
def get_forex_data(
pair: str = "EUR_USD",
tf: str = "1h",
years_back: int = 2,
source: str = "yahoo",
cache: bool = True,
) -> pd.DataFrame:
"""
Get forex OHLC data.
Parameters
----------
pair : str
e.g. "EUR_USD", "GBP_USD", "USD_JPY"
tf : str
Timeframe: 1m, 5m, 15m, 30m, 1h, 4h, 1d
For Dukascopy: M1, M5, ..., D1
years_back : int
How many years of history
source : str
"yahoo" (default, free, no key) | "oanda" | "dukascopy"
cache : bool
Cache to parquet files
"""
if source == "oanda":
try:
return get_oanda_candles(pair, tf.replace("1h", "H1").upper(), 5000)
except Exception as e:
print(f"OANDA error: {e}. Falling back to yahoo.")
source = "yahoo"
if source == "dukascopy":
now = datetime.now(timezone.utc)
start = now - timedelta(days=365 * years_back)
# Map timeframe
tf_map_rev = {v: k for k, v in TF_MAP.items()}
duka_tf = tf_map_rev.get(int(tf.replace("h", "")) * 3600 if "h" in tf else
int(tf.replace("m", "")) * 60 if "m" in tf else 86400, "H1")
# Download monthly, resample later if needed
all_dfs = []
y, m = start.year, start.month
while (y, m) <= (now.year, now.month):
df = get_dukascopy_month(pair, y, m, cache=cache)
if not df.empty:
all_dfs.append(df)
m += 1
if m > 12:
m = 1; y += 1
if not all_dfs:
return pd.DataFrame()
df = pd.concat(all_dfs, ignore_index=True)
df = df.drop_duplicates(subset=["time"]).sort_values("time")
df = df[(df["time"] >= pd.Timestamp(start)) & (df["time"] <= pd.Timestamp(now))]
# Resample if not M1
if duka_tf != "M1" and duka_tf in TF_MAP:
rule = f"{TF_MAP[duka_tf]}s"
df = df.set_index("time").resample(rule).agg({
"open": "first", "high": "max", "low": "min",
"close": "last", "volume": "sum", "pair": "last",
}).dropna().reset_index()
return df
# Default: Yahoo
return get_yahoo_data(pair, tf, years_back, cache=cache)
# ──────────────────────────────────────────────
# Available pairs & check
# ──────────────────────────────────────────────
AVAILABLE_PAIRS = list(YAHOO_PAIRS.keys())
def list_pairs():
"""Print all available forex pairs."""
print("Available forex pairs:")
for p in AVAILABLE_PAIRS:
print(f" • {p}")
return AVAILABLE_PAIRS
# ──────────────────────────────────────────────
# Quick test
# ──────────────────────────────────────────────
if __name__ == "__main__":
print("Testing Yahoo Finance forex downloader...")
df = get_forex_data("EUR_USD", "1d", years_back=1)
if not df.empty:
print(f"✅ EUR/USD Daily: {len(df):,} candles")
print(f" Range: {df['time'].min():%Y-%m-%d} → {df['time'].max():%Y-%m-%d}")
print(f" Latest: {df[['time', 'close', 'volume']].tail(3).to_string(index=False)}")
else:
print("⚠️ No data. Trying Dukascopy as fallback...")
df = get_forex_data("EUR_USD", "D1", years_back=1, source="dukascopy")
if not df.empty:
print(f"✅ Dukascopy EUR/USD: {len(df):,} candles")
else:
print("❌ Both sources failed. Check internet or use OANDA.")
forex-quant.code-workspace
+26
View File
@@ -0,0 +1,26 @@
{
"name": "Forex Quant Trading",
"folders": [
{
"path": "/Users/addysmacmini/forex-quant"
}
],
"settings": {
"python.defaultInterpreterPath": "/Users/addysmacmini/forex-quant/.venv/bin/python",
"python.terminal.activateEnvironment": true,
"files.exclude": {
"**/__pycache__": true,
"**/*.pyc": true,
".venv": true
},
"notebook.lineNumbers": "on",
"jupyter.interactiveWindow.creationMode": "perFile"
},
"extensions": {
"recommendations": [
"ms-python.python",
"ms-toolsai.jupyter",
"ms-python.vscode-pylance"
]
}
}
logs/multi_1d_2yr.csv
+5
View File
@@ -0,0 +1,5 @@
pair,tf,total_return,buy_hold_return,sharpe,max_dd,win_rate,trades,exposure_pct
EUR_USD,1d,11.97,9.01,0.97,-4.85,0.0,1,59.4
USD_JPY,1d,6.14,1.14,0.57,-4.23,0.0,1,39.5
GBP_USD,1d,1.79,7.97,0.16,-9.36,0.0,1,81.8
USD_CHF,1d,-10.78,-13.8,-0.71,-16.75,0.0,1,75.2
1 pair tf total_return buy_hold_return sharpe max_dd win_rate trades exposure_pct
2 EUR_USD 1d 11.97 9.01 0.97 -4.85 0.0 1 59.4
3 USD_JPY 1d 6.14 1.14 0.57 -4.23 0.0 1 39.5
4 GBP_USD 1d 1.79 7.97 0.16 -9.36 0.0 1 81.8
5 USD_CHF 1d -10.78 -13.8 -0.71 -16.75 0.0 1 75.2
strategies/momentum.py
+271
View File
@@ -0,0 +1,271 @@
"""
Momentum + Volatility Filter Strategy
Core logic:
- Buy when short-term MA crosses above medium-term MA
- MACD momentum confirmation
- Volatility filter via ATR
- RSI overbought/oversold exit
Works with OR without TA-Lib (uses pandas rolling if TA-Lib unavailable).
"""
import sys
from pathlib import Path
import numpy as np
import pandas as pd
sys.path.insert(0, str(Path(__file__).parent.parent))
# Try TA-Lib, fall back to pandas implementation
try:
import talib
HAS_TALIB = True
except ImportError:
HAS_TALIB = False
def _sma(values, period):
"""Simple Moving Average."""
if HAS_TALIB:
return talib.SMA(values, timeperiod=period)
return pd.Series(values).rolling(period).mean().values
def _ema(values, period):
"""Exponential Moving Average."""
if HAS_TALIB:
return talib.EMA(values, timeperiod=period)
return pd.Series(values).ewm(span=period, adjust=False).mean().values
def _rsi(values, period=14):
"""Relative Strength Index."""
if HAS_TALIB:
return talib.RSI(values, timeperiod=period)
series = pd.Series(values)
delta = series.diff()
gain = delta.where(delta > 0, 0).rolling(period).mean()
loss = (-delta.where(delta < 0, 0)).rolling(period).mean()
rs = gain / loss.replace(0, np.nan)
rsi = 100 - (100 / (1 + rs))
return rsi.values
def _macd(values, fast=12, slow=26, signal=9):
"""MACD."""
if HAS_TALIB:
macd, macd_signal, macd_hist = talib.MACD(values, fast, slow, signal)
return macd, macd_signal, macd_hist
ema_fast = _ema(values, fast)
ema_slow = _ema(values, slow)
macd = ema_fast - ema_slow
signal_line = _ema(macd, signal)
hist = macd - signal_line
return macd, signal_line, hist
def _atr(high, low, close, period=14):
"""Average True Range."""
if HAS_TALIB:
return talib.ATR(high, low, close, timeperiod=period)
high, low, close = pd.Series(high), pd.Series(low), pd.Series(close)
tr = pd.concat([
high - low,
(high - close.shift()).abs(),
(low - close.shift()).abs(),
], axis=1).max(axis=1)
return tr.rolling(period).mean().values
def _bbands(values, period=20, nbdev=2):
"""Bollinger Bands."""
if HAS_TALIB:
return talib.BBANDS(values, timeperiod=period, nbdevup=nbdev, nbdevdn=nbdev)
series = pd.Series(values)
sma = series.rolling(period).mean()
std = series.rolling(period).std()
upper = sma + nbdev * std
lower = sma - nbdev * std
return upper.values, sma.values, lower.values
def add_indicators(df: pd.DataFrame) -> pd.DataFrame:
"""Add technical indicators to OHLC DataFrame."""
df = df.copy()
close = df["close"].values
high = df["high"].values
low = df["low"].values
volume = df["volume"].values.astype(np.float64)
# Moving averages
df["ma_fast"] = _sma(close, 8)
df["ma_mid"] = _sma(close, 21)
df["ma_slow"] = _sma(close, 50)
# EMA
df["ema_fast"] = _ema(close, 12)
df["ema_slow"] = _ema(close, 26)
# MACD
macd, macd_signal, macd_hist = _macd(close, 12, 26, 9)
df["macd"] = macd
df["macd_signal"] = macd_signal
df["macd_hist"] = macd_hist
# RSI
df["rsi"] = _rsi(close, 14)
# ATR
df["atr"] = _atr(high, low, close, 14)
df["atr_pct"] = df["atr"] / close * 100
# Bollinger Bands
upper, mid, lower = _bbands(close, 20, 2)
df["bb_upper"] = upper
df["bb_mid"] = mid
df["bb_lower"] = lower
df["bb_width"] = (upper - lower) / mid * 100
return df
def generate_signals(
df: pd.DataFrame,
atr_min_pct: float = 0.05,
atr_max_pct: float = 1.0,
rsi_oversold: float = 30.0,
rsi_overbought: float = 70.0,
use_macd_filter: bool = True,
) -> pd.DataFrame:
"""
Generate trading signals from indicators.
Returns df with added 'signal' column: 1 = long, -1 = short, 0 = flat.
"""
df = df.copy()
df["signal"] = 0
if len(df) < 60:
return df
# Core momentum entry conditions
bull_trend = df["ma_fast"] > df["ma_mid"]
macd_bull = (df["macd_hist"] > 0) & (df["macd_hist"].shift(1) <= 0)
price_strong = (
(df["close"] > df["ma_fast"]) &
(df["close"] > df["ma_mid"]) &
(df["close"] > df["ma_slow"])
)
# Core momentum exit
bear_trend = df["ma_fast"] < df["ma_mid"]
macd_bear = (df["macd_hist"] < 0) & (df["macd_hist"].shift(1) >= 0)
# Volatility filter
valid_vol = (df["atr_pct"] >= atr_min_pct) & (df["atr_pct"] <= atr_max_pct)
# RSI filter
rsi_not_overbought = df["rsi"] < rsi_overbought
# Long entry
long_entry = bull_trend & valid_vol & price_strong & rsi_not_overbought
if use_macd_filter:
long_entry = long_entry & macd_bull
# Long exit
long_exit = bear_trend | macd_bear | (df["rsi"] > rsi_overbought + 10)
# Apply signals
df.loc[long_entry, "signal"] = 1
df.loc[long_exit & (df["signal"].shift(1) == 1), "signal"] = 0
# Forward-fill (hold between entries/exits)
df["position"] = df["signal"].replace(0, np.nan).ffill().fillna(0)
return df
def calculate_performance(df: pd.DataFrame, spread_cost: float = 0.0001) -> dict:
"""
Calculate basic strategy metrics.
spread_cost = estimated spread + commission in price units (~1 pip for EUR/USD)
"""
df = df.copy()
df["returns"] = df["close"].pct_change()
df["strategy_returns"] = df["position"].shift(1) * df["returns"]
# Subtract transaction costs on signal changes
df["trades"] = df["position"].diff().abs().clip(0)
df["strategy_returns"] -= df["trades"] * spread_cost / df["close"]
total_return = (1 + df["strategy_returns"]).prod() - 1
buy_hold_return = (1 + df["returns"]).prod() - 1
sharpe = np.nan
if df["strategy_returns"].std() > 0:
sharpe = (
df["strategy_returns"].mean()
/ df["strategy_returns"].std()
* np.sqrt(252)
)
max_drawdown = _max_drawdown((1 + df["strategy_returns"]).cumprod())
# Count actual trades (entry = position change from 0 to 1)
pos = df["position"].values
entries = np.where((pos[1:] == 1) & (pos[:-1] == 0))[0]
num_trades = len(entries)
win_rate = np.nan
if num_trades > 0:
trade_returns = []
for entry_idx in entries:
# Find exit after this entry
exit_idx = np.where((pos[entry_idx + 1:] == 0))[0]
if len(exit_idx) > 0:
exit_idx = entry_idx + 1 + exit_idx[0]
trade_return = df["close"].iloc[exit_idx] / df["close"].iloc[entry_idx] - 1
trade_returns.append(trade_return)
else:
trade_returns.append(df["close"].iloc[-1] / df["close"].iloc[entry_idx] - 1)
if trade_returns:
win_rate = sum(1 for r in trade_returns if r > 0) / len(trade_returns)
return {
"total_return_pct": total_return * 100,
"buy_hold_return_pct": buy_hold_return * 100,
"sharpe_ratio": round(sharpe, 2),
"max_drawdown_pct": max_drawdown * 100,
"win_rate_pct": win_rate * 100 if not np.isnan(win_rate) else 0,
"num_trades": num_trades,
"exposure_pct": (df["position"] != 0).mean() * 100,
}
def _max_drawdown(equity_curve: pd.Series) -> float:
peak = equity_curve.expanding().max()
dd = (equity_curve - peak) / peak
return min(dd.min(), 0)
# ──────────────────────────────────────────────
# Quick test
# ──────────────────────────────────────────────
if __name__ == "__main__":
from data.fx_data import get_forex_data
print(f"TA-Lib: {'✅ enabled' if HAS_TALIB else '❌ not available (using pandas fallback)'}")
print("Loading EUR/USD daily data...")
df = get_forex_data("EUR_USD", "1d", years_back=2)
if df.empty:
print("No data loaded.")
exit(1)
df = add_indicators(df)
df = generate_signals(df)
perf = calculate_performance(df)
print("\n📊 Strategy Performance (EUR/USD Daily, 2yr)")
for k, v in perf.items():
print(f" {k}: {v}")