-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy path“ProfitSight”
More file actions
111 lines (72 loc) · 5.51 KB
/
Copy path“ProfitSight”
File metadata and controls
111 lines (72 loc) · 5.51 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
=====================
backtest/metrics.py (expanded for ProfitSight)
=====================
from typing import List, Dict import math
def cumulative_returns(equity_curve: List[Dict]) -> float: if not equity_curve: return 0.0 start = equity_curve[0]["equity"] end = equity_curve[-1]["equity"] return (end / start) - 1.0 if start != 0 else 0.0
def max_drawdown(equity_curve: List[Dict]) -> float: peak = -math.inf max_dd = 0.0 for point in equity_curve: val = point["equity"] if val > peak: peak = val dd = (peak - val) / peak if peak != 0 else 0 if dd > max_dd: max_dd = dd return max_dd
def sharpe_ratio(equity_curve: List[Dict], risk_free_rate: float = 0.0) -> float: returns = [] for i in range(1, len(equity_curve)): prev = equity_curve[i - 1]["equity"] cur = equity_curve[i]["equity"] if prev == 0: returns.append(0.0) else: returns.append((cur / prev) - 1) if len(returns) < 2: return 0.0 mean_ret = sum(returns) / len(returns) variance = sum((r - mean_ret) ** 2 for r in returns) / (len(returns) - 1) std = math.sqrt(variance) if variance > 0 else 0.0 if std == 0: return 0.0 sharpe = (mean_ret * 252 - risk_free_rate) / (std * (252 ** 0.5)) return sharpe
def per_trade_stats(trade_log: List[Dict]) -> Dict: wins = [] losses = [] for i in range(1, len(trade_log), 2): # BUY followed by SELL buy = trade_log[i - 1] sell = trade_log[i] if buy['action'] != 'BUY' or sell['action'] != 'SELL': continue pnl = (sell['price'] - buy['price']) * buy['shares'] - buy['fee'] - sell['fee'] if pnl > 0: wins.append(pnl) else: losses.append(pnl) total_trades = len(wins) + len(losses) win_rate = len(wins)/total_trades if total_trades > 0 else 0.0 avg_win = sum(wins)/len(wins) if wins else 0.0 avg_loss = sum(losses)/len(losses) if losses else 0.0 expectancy = (win_rate * avg_win + (1 - win_rate) * avg_loss) if total_trades > 0 else 0.0 return { 'total_trades': total_trades, 'win_rate': win_rate, 'avg_win': avg_win, 'avg_loss': avg_loss, 'expectancy': expectancy }
=====================
scripts/backtest_all.py
=====================
import argparse from main import run_pipeline from backtest.backtester import Backtester from backtest.metrics import cumulative_returns, max_drawdown, sharpe_ratio, per_trade_stats
import pandas as pd
def main(): parser = argparse.ArgumentParser(description='Run backtests for multiple configurations.') parser.add_argument('--symbols', nargs='+', default=['BTC-USD'], help='List of symbols to backtest') parser.add_argument('--history', type=int, default=365, help='History length') parser.add_argument('--horizon', type=int, default=7, help='Forecast horizon') parser.add_argument('--capital', type=float, default=10000.0, help='Starting capital') args = parser.parse_args()
all_results = []
for sym in args.symbols:
out = run_pipeline(symbol=sym, days_history=args.history, forecast_horizon=args.horizon, capital=args.capital)
trade_res = out.get('trade_result', {})
metrics = {
'cumulative_returns': cumulative_returns(trade_res.get('equity_curve', [])),
'max_drawdown': max_drawdown(trade_res.get('equity_curve', [])),
'sharpe': sharpe_ratio(trade_res.get('equity_curve', [])),
}
trade_stats = per_trade_stats(trade_res.get('trade_log', []))
all_results.append({'symbol': sym, 'metrics': metrics, 'trade_stats': trade_stats})
df = pd.DataFrame(all_results)
print(df)
if name == 'main': main()
=====================
dashboard/app.py (updated overlay)
=====================
import streamlit as st import pandas as pd import plotly.graph_objs as go from main import run_pipeline from data.loader import fetch_historical
st.set_page_config(page_title='ProfitSight Dashboard', layout='wide') st.title('ProfitSight — Historical + Forecast Overlay')
with st.sidebar: symbol = st.text_input('Symbol', value='BTC-USD') days_history = st.number_input('History days', min_value=60, max_value=1825, value=365) horizon = st.number_input('Forecast horizon (days)', min_value=1, max_value=60, value=7) capital = st.number_input('Capital', min_value=100.0, value=10000.0) run_btn = st.button('Run Pipeline')
if run_btn: with st.spinner('Fetching historical data and running pipeline...'): hist_df = fetch_historical(symbol, days=days_history) out = run_pipeline(symbol=symbol, days_history=days_history, forecast_horizon=horizon, capital=capital)
preds = out.get('predictions', [])
trade_res = out.get('trade_result', {})
# Historical prices
hist_df['Date'] = pd.to_datetime(hist_df['Date'])
fig = go.Figure()
fig.add_trace(go.Scatter(x=hist_df['Date'], y=hist_df['Close'], mode='lines', name='Historical Close'))
# Forecast overlay
future_dates = pd.date_range(start=hist_df['Date'].iloc[-1] + pd.Timedelta(days=1), periods=len(preds))
fig.add_trace(go.Scatter(x=future_dates, y=preds, mode='lines+markers', name='Predicted'))
st.plotly_chart(fig, use_container_width=True)
st.subheader('Trades')
st.write(trade_res.get('trade_log', []))
st.subheader('Metrics & Evaluation')
st.json(out.get('evaluation', {}))
=====================
.github/workflows/ci.yml
=====================
name: Python CI
on: push: branches: [ main ] pull_request: branches: [ main ]
jobs: build-test: runs-on: ubuntu-latest
strategy:
matrix:
python-version: [3.10, 3.11]
steps:
- uses: actions/checkout@v3
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install pytest
- name: Run tests
run: |
pytest --maxfail=1 --disable-warnings -v