sami7777/biggfish
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote main: resolve conflicts keeping scalping strategy changes

Browse Source 
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
sami7777
2026-03-12 16:46:28 -07:00
parent 82c68fa8b0 6a35ba0f0f
commit 62822a8675
11 changed files with 2948 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.claude/settings.local.json
+1 -11
View File
@@ -3,17 +3,7 @@
"allow": [
"Bash(python -c:*)",
"Bash(ssh:*)",
"Bash(scp:*)",
"WebFetch(domain:docs.alpaca.markets)",
"Bash(curl:*)",
"Bash(python -m json.tool:*)",
"Bash(python3:*)",
"WebFetch(domain:sag.sh)",
"WebFetch(domain:summarize.sh)",
"Bash(git init:*)",
"Bash(git remote add:*)",
"Bash(git remote set-url:*)",
"Bash(git add:*)"
"Bash(scp:*)"
]
}
}
.gitignore
+16 -24
View File
@@ -1,6 +1,3 @@
# Configuration
config/config.json
# Python
__pycache__/
*.py[cod]
@@ -10,30 +7,24 @@ __pycache__/
env/
venv/
ENV/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
.eggs/
# Data & Models (root level only)
/data/
*.db
*.db-journal
*.pkl
*.h5
/models/
# Logs
logs/
*.log
# Data
data/
*.db
*.sqlite
# Config (keep examples)
config/auto_config.json
config/auto_config.json.backup*
# IDE
.vscode/
@@ -46,6 +37,7 @@ data/
.DS_Store
Thumbs.db
# Environment
.env
.env.local
# Testing
.pytest_cache/
.coverage
htmlcov/
src/data/__init__.py
View File
src/data/candle_cache.py
+174
View File
@@ -0,0 +1,174 @@
"""
OHLCV Candle Cache
Fetches historical data via yfinance (stocks) or OANDA (forex), caches in SQLite.
"""
import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta
from typing import List, Optional
from loguru import logger
class CandleCache:
"""Fetches OHLCV data and caches in SQLite"""
def __init__(self, store, oanda_broker=None):
self.store = store
self.oanda = oanda_broker
def _is_forex(self, symbol: str) -> bool:
"""Check if symbol is a forex pair (OANDA format: XXX_YYY)"""
return '_' in symbol and len(symbol) == 7
def fetch_and_cache(self, symbol: str, timeframe: str = '1h',
lookback_days: int = 90) -> Optional[pd.DataFrame]:
"""
Fetch candles, store in DB, return DataFrame.
Routes to OANDA for forex pairs, yfinance for stocks.
"""
if self._is_forex(symbol):
return self._fetch_oanda(symbol, timeframe, lookback_days)
else:
return self._fetch_yfinance(symbol, timeframe, lookback_days)
def _fetch_oanda(self, symbol: str, timeframe: str,
lookback_days: int) -> Optional[pd.DataFrame]:
"""Fetch forex data from OANDA API"""
if not self.oanda:
logger.debug(f"No OANDA broker configured, skipping {symbol}")
return self.store.get_candles(symbol, timeframe)
# Check what we already have
latest_ts = self.store.get_latest_candle_timestamp(symbol, timeframe)
if latest_ts:
latest_dt = datetime.utcfromtimestamp(latest_ts / 1000)
since_dt = latest_dt + timedelta(minutes=1)
logger.debug(f"Cache has data until {latest_dt} for {symbol}/{timeframe}")
else:
since_dt = datetime.utcnow() - timedelta(days=lookback_days)
end_dt = datetime.utcnow()
if since_dt >= end_dt - timedelta(minutes=5):
logger.debug(f"Cache is up to date for {symbol}/{timeframe}")
return self.store.get_candles(symbol, timeframe)
try:
bars = self.oanda.fetch_bars_range(
symbol, timeframe=timeframe, start=since_dt, end=end_dt
)
if bars:
self.store.store_candles(symbol, timeframe, bars)
logger.info(f"Cached {len(bars)} candles for {symbol}/{timeframe} (OANDA)")
if timeframe == '4h':
return self._resample_to_4h(symbol)
return self.store.get_candles(symbol, timeframe)
except Exception as e:
logger.error(f"Error fetching OANDA candles for {symbol}: {e}")
return self.store.get_candles(symbol, timeframe)
def _fetch_yfinance(self, symbol: str, timeframe: str,
lookback_days: int) -> Optional[pd.DataFrame]:
"""Fetch stock data from yfinance"""
tf_map = {
'1m': '1m', '5m': '5m', '15m': '15m',
'1h': '1h', '4h': '1h', '1d': '1d'
}
yf_interval = tf_map.get(timeframe, '1h')
max_lookback = {
'1m': 7, '5m': 60, '15m': 60, '1h': 730, '1d': 3650
}
lookback_days = min(lookback_days, max_lookback.get(yf_interval, 90))
latest_ts = self.store.get_latest_candle_timestamp(symbol, timeframe)
if latest_ts:
latest_dt = datetime.utcfromtimestamp(latest_ts / 1000)
since_dt = latest_dt + timedelta(minutes=1)
logger.debug(f"Cache has data until {latest_dt} for {symbol}/{timeframe}")
else:
since_dt = datetime.utcnow() - timedelta(days=lookback_days)
try:
end_dt = datetime.utcnow()
if since_dt >= end_dt - timedelta(minutes=5):
logger.debug(f"Cache is up to date for {symbol}/{timeframe}")
return self.store.get_candles(symbol, timeframe)
ticker = yf.Ticker(symbol)
hist = ticker.history(
start=since_dt.strftime('%Y-%m-%d'),
end=end_dt.strftime('%Y-%m-%d'),
interval=yf_interval
)
if hist.empty:
logger.debug(f"No new data for {symbol}/{timeframe}")
return self.store.get_candles(symbol, timeframe)
candles = []
for ts, row in hist.iterrows():
candles.append({
'timestamp': int(ts.timestamp() * 1000),
'open': float(row['Open']),
'high': float(row['High']),
'low': float(row['Low']),
'close': float(row['Close']),
'volume': float(row['Volume'])
})
if candles:
self.store.store_candles(symbol, timeframe, candles)
logger.info(f"Cached {len(candles)} candles for {symbol}/{timeframe}")
if timeframe == '4h':
return self._resample_to_4h(symbol)
return self.store.get_candles(symbol, timeframe)
except Exception as e:
logger.error(f"Error fetching candles for {symbol}: {e}")
return self.store.get_candles(symbol, timeframe)
def _resample_to_4h(self, symbol: str) -> Optional[pd.DataFrame]:
"""Resample 1h candles to 4h"""
df = self.store.get_candles(symbol, '1h')
if df is None or df.empty:
return None
resampled = df.resample('4h').agg({
'open': 'first',
'high': 'max',
'low': 'min',
'close': 'last',
'volume': 'sum'
}).dropna()
return resampled
def warm_cache(self, symbols: List[str], timeframes: List[str],
lookback_days: int = 90):
"""Pre-fetch historical data for all symbols/timeframes"""
total = len(symbols) * len(timeframes)
done = 0
for symbol in symbols:
for tf in timeframes:
self.fetch_and_cache(symbol, tf, lookback_days)
done += 1
if done % 5 == 0:
logger.info(f"Cache warmup: {done}/{total} complete")
logger.info(f"Cache warmup complete: {total} symbol/timeframe combinations")
def update_cache(self, symbols: List[str], timeframes: List[str]):
"""Incremental update - fetch only new candles since last cached"""
for symbol in symbols:
for tf in timeframes:
self.fetch_and_cache(symbol, tf, lookback_days=2)
def get_cached(self, symbol: str, timeframe: str,
start: datetime = None, end: datetime = None) -> Optional[pd.DataFrame]:
"""Get cached candles as DataFrame"""
return self.store.get_candles(symbol, timeframe, start, end)
src/data/features.py
+175
View File
@@ -0,0 +1,175 @@
"""
Feature Engineering Pipeline
Computes technical indicators and normalizes features for ML input.
"""
import pandas as pd
import numpy as np
from loguru import logger
class FeatureEngine:
"""Computes technical indicators and normalizes for ML input"""
FEATURE_NAMES = [
# Trend (6)
'sma_10', 'sma_20', 'sma_50',
'ema_10', 'ema_20', 'ema_50',
# MACD (3)
'macd', 'macd_signal', 'macd_hist',
# Momentum (4)
'rsi_14', 'stoch_k', 'stoch_d', 'roc_10',
# Volatility (5)
'bb_upper', 'bb_middle', 'bb_lower', 'bb_width', 'atr_14',
# Volume (3)
'obv', 'volume_sma_20', 'volume_ratio',
# Price action (5)
'returns_1', 'returns_5', 'returns_10', 'returns_20',
'high_low_range',
# Relative position (3)
'price_vs_sma20', 'price_vs_sma50', 'atr_pct',
]
NUM_FEATURES = len(FEATURE_NAMES) # 29
def compute(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Compute all features from raw OHLCV DataFrame.
Input df must have columns: open, high, low, close, volume
Returns df with all feature columns appended, NaN rows dropped.
"""
df = df.copy()
c = df['close']
h = df['high']
l = df['low']
o = df['open']
v = df['volume']
# --- Trend indicators ---
df['sma_10'] = c.rolling(10).mean()
df['sma_20'] = c.rolling(20).mean()
df['sma_50'] = c.rolling(50).mean()
df['ema_10'] = c.ewm(span=10).mean()
df['ema_20'] = c.ewm(span=20).mean()
df['ema_50'] = c.ewm(span=50).mean()
# --- MACD ---
ema12 = c.ewm(span=12).mean()
ema26 = c.ewm(span=26).mean()
df['macd'] = ema12 - ema26
df['macd_signal'] = df['macd'].ewm(span=9).mean()
df['macd_hist'] = df['macd'] - df['macd_signal']
# --- Momentum ---
# RSI
delta = c.diff()
gain = delta.where(delta > 0, 0.0).rolling(14).mean()
loss = (-delta.where(delta < 0, 0.0)).rolling(14).mean()
rs = gain / loss.replace(0, np.nan)
df['rsi_14'] = 100 - (100 / (1 + rs))
# Stochastic
low14 = l.rolling(14).min()
high14 = h.rolling(14).max()
df['stoch_k'] = 100 * (c - low14) / (high14 - low14).replace(0, np.nan)
df['stoch_d'] = df['stoch_k'].rolling(3).mean()
# Rate of change
df['roc_10'] = c.pct_change(10) * 100
# --- Volatility ---
# Bollinger Bands
sma20 = c.rolling(20).mean()
std20 = c.rolling(20).std()
df['bb_upper'] = sma20 + 2 * std20
df['bb_middle'] = sma20
df['bb_lower'] = sma20 - 2 * std20
df['bb_width'] = (df['bb_upper'] - df['bb_lower']) / df['bb_middle'].replace(0, np.nan)
# ATR
tr = pd.concat([
h - l,
(h - c.shift(1)).abs(),
(l - c.shift(1)).abs()
], axis=1).max(axis=1)
df['atr_14'] = tr.rolling(14).mean()
# --- Volume ---
# OBV
obv = pd.Series(0.0, index=df.index)
obv_vals = [0.0]
for i in range(1, len(df)):
if c.iloc[i] > c.iloc[i-1]:
obv_vals.append(obv_vals[-1] + v.iloc[i])
elif c.iloc[i] < c.iloc[i-1]:
obv_vals.append(obv_vals[-1] - v.iloc[i])
else:
obv_vals.append(obv_vals[-1])
df['obv'] = obv_vals
df['volume_sma_20'] = v.rolling(20).mean()
df['volume_ratio'] = v / df['volume_sma_20'].replace(0, np.nan)
# --- Price action ---
df['returns_1'] = c.pct_change(1) * 100
df['returns_5'] = c.pct_change(5) * 100
df['returns_10'] = c.pct_change(10) * 100
df['returns_20'] = c.pct_change(20) * 100
df['high_low_range'] = (h - l) / c.replace(0, np.nan)
# --- Relative position ---
df['price_vs_sma20'] = (c - df['sma_20']) / df['sma_20'].replace(0, np.nan) * 100
df['price_vs_sma50'] = (c - df['sma_50']) / df['sma_50'].replace(0, np.nan) * 100
df['atr_pct'] = df['atr_14'] / c.replace(0, np.nan) * 100
# Drop NaN rows from indicator warmup
df.dropna(inplace=True)
return df
def normalize(self, df: pd.DataFrame, window: int = 200) -> pd.DataFrame:
"""
Z-score normalize feature columns using a rolling window.
Avoids look-ahead bias by using only past data.
"""
result = df.copy()
for col in self.FEATURE_NAMES:
if col in result.columns:
rolling_mean = result[col].rolling(window, min_periods=20).mean()
rolling_std = result[col].rolling(window, min_periods=20).std()
result[col] = (result[col] - rolling_mean) / rolling_std.replace(0, np.nan)
result.dropna(inplace=True)
# Clip extreme values
for col in self.FEATURE_NAMES:
if col in result.columns:
result[col] = result[col].clip(-3, 3)
return result
def get_state_vector(self, df: pd.DataFrame, index: int = -1) -> np.ndarray:
"""
Extract a single normalized state vector at a given index.
Returns shape: (NUM_FEATURES,)
"""
if index < 0:
index = len(df) + index
row = df.iloc[index]
features = []
for col in self.FEATURE_NAMES:
if col in df.columns:
val = row[col]
features.append(0.0 if pd.isna(val) else float(val))
else:
features.append(0.0)
return np.array(features, dtype=np.float32)
def compute_and_normalize(self, df: pd.DataFrame) -> pd.DataFrame:
"""Compute features and normalize in one step"""
featured = self.compute(df)
if len(featured) < 20:
return featured
return self.normalize(featured)
src/data/krystie-events.json
+504
View File
@@ -0,0 +1,504 @@
{
"events": [
{
"time": "2026-03-12T22:53:54Z",
"type": "ga_milestone",
"data": {
"generation": 50334,
"fitness": 23.8731
}
},
{
"time": "2026-03-12T22:55:44Z",
"type": "ga_milestone",
"data": {
"generation": 50349,
"fitness": 23.8731
}
},
{
"time": "2026-03-12T22:57:38Z",
"type": "ga_milestone",
"data": {
"generation": 50364,
"fitness": 23.8731
}
},
{
"time": "2026-03-12T22:59:44Z",
"type": "ga_milestone",
"data": {
"generation": 50379,
"fitness": 23.8731
}
},
{
"time": "2026-03-12T23:01:50Z",
"type": "ga_milestone",
"data": {
"generation": 50394,
"fitness": 23.8731
}
},
{
"time": "2026-03-12T23:03:52Z",
"type": "ga_milestone",
"data": {
"generation": 50409,
"fitness": 23.8731
}
},
{
"time": "2026-03-12T23:05:50Z",
"type": "ga_milestone",
"data": {
"generation": 50424,
"fitness": 23.8731
}
},
{
"time": "2026-03-12T23:07:42Z",
"type": "ga_milestone",
"data": {
"generation": 50439,
"fitness": 23.8731
}
},
{
"time": "2026-03-12T23:09:42Z",
"type": "ga_milestone",
"data": {
"generation": 50454,
"fitness": 23.8731
}
},
{
"time": "2026-03-12T23:11:47Z",
"type": "ga_milestone",
"data": {
"generation": 50469,
"fitness": 23.8731
}
},
{
"time": "2026-03-12T23:13:24Z",
"type": "bot_started",
"data": {
"message": "BIGGFISH autonomous trader started"
}
},
{
"time": "2026-03-12T23:14:39Z",
"type": "bot_started",
"data": {
"message": "BIGGFISH autonomous trader started"
}
},
{
"time": "2026-03-12T23:15:49Z",
"type": "trade_open",
"data": {
"symbol": "AUD_USD",
"side": "buy",
"amount": 27957,
"entry_price": 0.70788
}
},
{
"time": "2026-03-12T23:15:51Z",
"type": "trade_open",
"data": {
"symbol": "USD_CAD",
"side": "buy",
"amount": 14515,
"entry_price": 1.36334
}
},
{
"time": "2026-03-12T23:15:53Z",
"type": "trade_open",
"data": {
"symbol": "EUR_GBP",
"side": "buy",
"amount": 22932,
"entry_price": 0.86296
}
},
{
"time": "2026-03-12T23:15:54Z",
"type": "trade_open",
"data": {
"symbol": "USD_CHF",
"side": "buy",
"amount": 25186,
"entry_price": 0.78568
}
},
{
"time": "2026-03-12T23:15:56Z",
"type": "trade_open",
"data": {
"symbol": "NZD_USD",
"side": "buy",
"amount": 33817,
"entry_price": 0.58514
}
},
{
"time": "2026-03-12T23:16:49Z",
"type": "ga_milestone",
"data": {
"generation": 15,
"fitness": 1.7479
}
},
{
"time": "2026-03-12T23:16:51Z",
"type": "daily_report",
"data": {
"equity": 98929.1,
"day_pnl": -1070.9,
"trades_count": 5
}
},
{
"time": "2026-03-12T23:17:44Z",
"type": "trade_open",
"data": {
"symbol": "GBP_USD",
"side": "buy",
"amount": 14823,
"entry_price": 1.3348
}
},
{
"time": "2026-03-12T23:17:47Z",
"type": "trade_close",
"data": {
"symbol": "AUD_USD",
"side": "sell",
"entry_price": null,
"exit_price": 0.70784,
"pnl": -0.56,
"pnl_pct": 0,
"exit_reason": "signal"
}
},
{
"time": "2026-03-12T23:18:12Z",
"type": "ga_milestone",
"data": {
"generation": 30,
"fitness": 1.8512
}
},
{
"time": "2026-03-12T23:19:43Z",
"type": "trade_open",
"data": {
"symbol": "EUR_USD",
"side": "buy",
"amount": 17179,
"entry_price": 1.1516
}
},
{
"time": "2026-03-12T23:19:48Z",
"type": "trade_open",
"data": {
"symbol": "USD_JPY",
"side": "buy",
"amount": 124,
"entry_price": 159.342
}
},
{
"time": "2026-03-12T23:19:51Z",
"type": "trade_close",
"data": {
"symbol": "USD_CAD",
"side": "sell",
"entry_price": null,
"exit_price": 1.3636,
"pnl": 1.89,
"pnl_pct": 0,
"exit_reason": "signal"
}
},
{
"time": "2026-03-12T23:19:56Z",
"type": "trade_close",
"data": {
"symbol": "NZD_USD",
"side": "sell",
"entry_price": null,
"exit_price": 0.585,
"pnl": -2.37,
"pnl_pct": 0,
"exit_reason": "signal"
}
},
{
"time": "2026-03-12T23:20:16Z",
"type": "ga_milestone",
"data": {
"generation": 45,
"fitness": 1.8512
}
},
{
"time": "2026-03-12T23:21:44Z",
"type": "trade_close",
"data": {
"symbol": "GBP_USD",
"side": "sell",
"entry_price": null,
"exit_price": 1.33478,
"pnl": -0.3,
"pnl_pct": -0.0,
"exit_reason": "signal"
}
},
{
"time": "2026-03-12T23:21:47Z",
"type": "trade_close",
"data": {
"symbol": "USD_CAD",
"side": "sell",
"entry_price": null,
"exit_price": 1.36348,
"pnl": 0.51,
"pnl_pct": 0,
"exit_reason": "signal"
}
},
{
"time": "2026-03-12T23:21:49Z",
"type": "trade_close",
"data": {
"symbol": "EUR_GBP",
"side": "sell",
"entry_price": null,
"exit_price": 0.86288,
"pnl": -0.92,
"pnl_pct": 0,
"exit_reason": "signal"
}
},
{
"time": "2026-03-12T23:22:12Z",
"type": "ga_milestone",
"data": {
"generation": 60,
"fitness": 1.8512
}
},
{
"time": "2026-03-12T23:23:49Z",
"type": "trade_close",
"data": {
"symbol": "USD_CAD",
"side": "sell",
"entry_price": null,
"exit_price": 1.36355,
"pnl": 0.76,
"pnl_pct": 0.02,
"exit_reason": "signal"
}
},
{
"time": "2026-03-12T23:23:53Z",
"type": "trade_close",
"data": {
"symbol": "USD_CHF",
"side": "sell",
"entry_price": null,
"exit_price": 0.78586,
"pnl": 2.27,
"pnl_pct": 0,
"exit_reason": "signal"
}
},
{
"time": "2026-03-12T23:24:15Z",
"type": "ga_milestone",
"data": {
"generation": 75,
"fitness": 1.8512
}
},
{
"time": "2026-03-12T23:25:45Z",
"type": "trade_open",
"data": {
"symbol": "GBP_USD",
"side": "buy",
"amount": 14822,
"entry_price": 1.33468
}
},
{
"time": "2026-03-12T23:25:51Z",
"type": "trade_close",
"data": {
"symbol": "EUR_GBP",
"side": "sell",
"entry_price": null,
"exit_price": 0.8628,
"pnl": -0.92,
"pnl_pct": 0,
"exit_reason": "signal"
}
},
{
"time": "2026-03-12T23:26:10Z",
"type": "ga_milestone",
"data": {
"generation": 90,
"fitness": 1.8512
}
},
{
"time": "2026-03-12T23:27:45Z",
"type": "trade_close",
"data": {
"symbol": "USD_JPY",
"side": "sell",
"entry_price": null,
"exit_price": 159.35,
"pnl": 0.99,
"pnl_pct": 0.01,
"exit_reason": "signal"
}
},
{
"time": "2026-03-12T23:28:10Z",
"type": "ga_milestone",
"data": {
"generation": 105,
"fitness": 1.8512
}
},
{
"time": "2026-03-12T23:29:45Z",
"type": "trade_close",
"data": {
"symbol": "GBP_USD",
"side": "sell",
"entry_price": null,
"exit_price": 1.33488,
"pnl": 2.96,
"pnl_pct": 0.01,
"exit_reason": "signal"
}
},
{
"time": "2026-03-12T23:29:46Z",
"type": "trade_open",
"data": {
"symbol": "USD_JPY",
"side": "buy",
"amount": 124,
"entry_price": 159.326
}
},
{
"time": "2026-03-12T23:29:48Z",
"type": "trade_close",
"data": {
"symbol": "AUD_USD",
"side": "sell",
"entry_price": null,
"exit_price": 0.70756,
"pnl": -2.24,
"pnl_pct": 0,
"exit_reason": "signal"
}
},
{
"time": "2026-03-12T23:29:52Z",
"type": "trade_close",
"data": {
"symbol": "USD_CHF",
"side": "sell",
"entry_price": null,
"exit_price": 0.78578,
"pnl": 1.26,
"pnl_pct": 0.01,
"exit_reason": "signal"
}
},
{
"time": "2026-03-12T23:29:54Z",
"type": "trade_close",
"data": {
"symbol": "NZD_USD",
"side": "sell",
"entry_price": null,
"exit_price": 0.58492,
"pnl": -3.72,
"pnl_pct": -0.04,
"exit_reason": "signal"
}
},
{
"time": "2026-03-12T23:30:25Z",
"type": "ga_milestone",
"data": {
"generation": 120,
"fitness": 1.8615
}
},
{
"time": "2026-03-12T23:31:43Z",
"type": "trade_close",
"data": {
"symbol": "EUR_USD",
"side": "sell",
"entry_price": null,
"exit_price": 1.15166,
"pnl": 0.52,
"pnl_pct": 0,
"exit_reason": "signal"
}
},
{
"time": "2026-03-12T23:31:45Z",
"type": "trade_open",
"data": {
"symbol": "GBP_USD",
"side": "buy",
"amount": 14821,
"entry_price": 1.33481
}
},
{
"time": "2026-03-12T23:31:48Z",
"type": "trade_open",
"data": {
"symbol": "USD_CAD",
"side": "buy",
"amount": 14506,
"entry_price": 1.36368
}
},
{
"time": "2026-03-12T23:31:51Z",
"type": "trade_open",
"data": {
"symbol": "USD_CHF",
"side": "buy",
"amount": 25174,
"entry_price": 0.78579
}
},
{
"time": "2026-03-12T23:32:10Z",
"type": "ga_milestone",
"data": {
"generation": 135,
"fitness": 1.8615
}
}
]
}
src/data/krystie-status.json
+114
View File
@@ -0,0 +1,114 @@
{
"updated_at": "2026-03-12T23:32:11Z",
"uptime_hours": 0.3,
"markets": {
"stocks": "CLOSED",
"forex": "OPEN"
},
"portfolio": {
"equity": 98905.3095,
"cash": 98924.0431,
"buying_power": 96190.7269,
"portfolio_value": 98905.3095,
"long_market_value": -18.733599999992293,
"day_pnl": -1094.578999999998,
"day_pnl_pct": -1.094578999999998
},
"positions": [
{
"symbol": "NZD_USD",
"qty": 16,
"entry_price": 0.58526,
"current_price": 0.58482875,
"unrealized_pnl": -0.0069
},
{
"symbol": "AUD_USD",
"qty": 7018,
"entry_price": 0.70794,
"current_price": 0.7074600056996295,
"unrealized_pnl": -3.3686
},
{
"symbol": "USD_JPY",
"qty": 140,
"entry_price": 159.338,
"current_price": 159.33775642857142,
"unrealized_pnl": -0.0341
},
{
"symbol": "USD_CHF",
"qty": 25181,
"entry_price": 0.78588,
"current_price": 0.7855858536197927,
"unrealized_pnl": -7.4069
},
{
"symbol": "GBP_USD",
"qty": 14839,
"entry_price": 1.3349,
"current_price": 1.3347701260192735,
"unrealized_pnl": -1.9272
},
{
"symbol": "USD_CAD",
"qty": 14526,
"entry_price": 1.36378,
"current_price": 1.3635518642434257,
"unrealized_pnl": -3.3139
},
{
"symbol": "EUR_GBP",
"qty": 5748,
"entry_price": 0.86301,
"current_price": 0.8626343562978428,
"unrealized_pnl": -2.1592
},
{
"symbol": "EUR_USD",
"qty": 8614,
"entry_price": 1.15168,
"current_price": 1.1516200046436034,
"unrealized_pnl": -0.5168
}
],
"learning": {
"ga_generation": 135,
"ga_best_fitness": 1.8615,
"rl_epsilon": 0.7471,
"rl_experiences": 710,
"rl_loss": 0.000668
},
"today_summary": {
"trades_count": 22,
"wins": 0,
"losses": 15,
"total_pnl": -239.08
},
"config": {
"stock_symbols": [
"SOUN",
"MARA",
"RIOT",
"BBAI",
"PLTR",
"HOOD",
"SOFI",
"COIN",
"RBLX",
"DKNG"
],
"forex_symbols": [
"EUR_USD",
"GBP_USD",
"USD_JPY",
"AUD_USD",
"USD_CAD",
"EUR_GBP",
"USD_CHF",
"NZD_USD"
],
"initial_capital": 100000,
"target_capital": 1000000
}
}
src/data/store.py
+389
View File
@@ -0,0 +1,389 @@
"""
SQLite Data Access Layer for BIGGFISH
Persists trades, candles, strategy performance, model checkpoints, and system state.
"""
import sqlite3
import json
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional
from loguru import logger
import pandas as pd
class DataStore:
"""SQLite data access layer for all BIGGFISH persistence"""
def __init__(self, db_path: str = "data/biggfish.db"):
self.db_path = Path(db_path)
self.db_path.parent.mkdir(parents=True, exist_ok=True)
self._conn = None
@property
def conn(self) -> sqlite3.Connection:
if self._conn is None:
self._conn = sqlite3.connect(str(self.db_path), timeout=30)
self._conn.row_factory = sqlite3.Row
self._conn.execute("PRAGMA journal_mode=WAL")
self._conn.execute("PRAGMA busy_timeout=5000")
return self._conn
def initialize(self):
"""Create all tables if they don't exist"""
c = self.conn
c.executescript("""
CREATE TABLE IF NOT EXISTS candles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
symbol TEXT NOT NULL,
timeframe TEXT NOT NULL,
timestamp INTEGER NOT NULL,
open REAL NOT NULL,
high REAL NOT NULL,
low REAL NOT NULL,
close REAL NOT NULL,
volume REAL NOT NULL,
UNIQUE(symbol, timeframe, timestamp)
);
CREATE INDEX IF NOT EXISTS idx_candles_lookup
ON candles(symbol, timeframe, timestamp);
CREATE TABLE IF NOT EXISTS trades (
id INTEGER PRIMARY KEY AUTOINCREMENT,
symbol TEXT NOT NULL,
side TEXT NOT NULL,
amount REAL NOT NULL,
entry_price REAL NOT NULL,
exit_price REAL,
entry_time TEXT NOT NULL,
exit_time TEXT,
strategy_id TEXT,
stop_loss REAL,
take_profit REAL,
pnl REAL,
pnl_pct REAL,
fees REAL DEFAULT 0,
order_id TEXT,
status TEXT DEFAULT 'open',
metadata TEXT
);
CREATE INDEX IF NOT EXISTS idx_trades_symbol ON trades(symbol, status);
CREATE INDEX IF NOT EXISTS idx_trades_strategy ON trades(strategy_id);
CREATE TABLE IF NOT EXISTS strategy_performance (
id INTEGER PRIMARY KEY AUTOINCREMENT,
strategy_id TEXT NOT NULL,
params TEXT NOT NULL,
backtest_start TEXT,
backtest_end TEXT,
total_trades INTEGER,
win_rate REAL,
profit_factor REAL,
sharpe_ratio REAL,
sortino_ratio REAL,
max_drawdown REAL,
total_return REAL,
avg_trade_pnl REAL,
recorded_at TEXT NOT NULL,
source TEXT DEFAULT 'backtest'
);
CREATE INDEX IF NOT EXISTS idx_strat_perf
ON strategy_performance(strategy_id, recorded_at);
CREATE TABLE IF NOT EXISTS model_checkpoints (
id INTEGER PRIMARY KEY AUTOINCREMENT,
model_name TEXT NOT NULL,
epoch INTEGER NOT NULL,
state_dict BLOB NOT NULL,
metrics TEXT,
saved_at TEXT NOT NULL,
UNIQUE(model_name, epoch)
);
CREATE TABLE IF NOT EXISTS evolution_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
generation INTEGER NOT NULL,
population TEXT NOT NULL,
best_fitness REAL NOT NULL,
avg_fitness REAL,
recorded_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS system_state (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
updated_at TEXT NOT NULL
);
""")
c.commit()
logger.info(f"Database initialized at {self.db_path}")
# --- Candle cache ---
def store_candles(self, symbol: str, timeframe: str, candles: List[Dict]):
"""Store OHLCV candles (upsert)"""
if not candles:
return
c = self.conn
c.executemany(
"""INSERT OR REPLACE INTO candles
(symbol, timeframe, timestamp, open, high, low, close, volume)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
[(symbol, timeframe, int(row['timestamp']),
row['open'], row['high'], row['low'], row['close'], row['volume'])
for row in candles]
)
c.commit()
def get_candles(self, symbol: str, timeframe: str,
start: datetime = None, end: datetime = None) -> Optional[pd.DataFrame]:
"""Return cached candles as DataFrame"""
query = "SELECT timestamp, open, high, low, close, volume FROM candles WHERE symbol=? AND timeframe=?"
params = [symbol, timeframe]
if start:
query += " AND timestamp >= ?"
params.append(int(start.timestamp() * 1000))
if end:
query += " AND timestamp <= ?"
params.append(int(end.timestamp() * 1000))
query += " ORDER BY timestamp ASC"
df = pd.read_sql_query(query, self.conn, params=params)
if df.empty:
return None
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
return df
def get_latest_candle_timestamp(self, symbol: str, timeframe: str) -> Optional[int]:
"""Get the most recent cached candle timestamp (epoch ms)"""
row = self.conn.execute(
"SELECT MAX(timestamp) as ts FROM candles WHERE symbol=? AND timeframe=?",
(symbol, timeframe)
).fetchone()
return row['ts'] if row and row['ts'] else None
def get_candle_count(self, symbol: str, timeframe: str) -> int:
"""Get number of cached candles"""
row = self.conn.execute(
"SELECT COUNT(*) as cnt FROM candles WHERE symbol=? AND timeframe=?",
(symbol, timeframe)
).fetchone()
return row['cnt'] if row else 0
# --- Trades ---
def record_trade(self, trade: Dict) -> int:
"""Record a trade, return its ID"""
c = self.conn
cursor = c.execute(
"""INSERT INTO trades
(symbol, side, amount, entry_price, exit_price, entry_time, exit_time,
strategy_id, stop_loss, take_profit, pnl, pnl_pct, fees, order_id, status, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(trade['symbol'], trade['side'], trade['amount'], trade['entry_price'],
trade.get('exit_price'), trade['entry_time'], trade.get('exit_time'),
trade.get('strategy_id'), trade.get('stop_loss'), trade.get('take_profit'),
trade.get('pnl'), trade.get('pnl_pct'), trade.get('fees', 0),
trade.get('order_id'), trade.get('status', 'open'),
json.dumps(trade.get('metadata', {})))
)
c.commit()
return cursor.lastrowid
def get_trades(self, symbol: str = None, status: str = None,
start: datetime = None, limit: int = 100) -> List[Dict]:
"""Get trades with optional filters"""
query = "SELECT * FROM trades WHERE 1=1"
params = []
if symbol:
query += " AND symbol=?"
params.append(symbol)
if status:
query += " AND status=?"
params.append(status)
if start:
query += " AND entry_time >= ?"
params.append(start.isoformat())
query += " ORDER BY entry_time DESC LIMIT ?"
params.append(limit)
rows = self.conn.execute(query, params).fetchall()
return [dict(r) for r in rows]
def get_open_positions(self) -> List[Dict]:
"""Get all open trades"""
rows = self.conn.execute(
"SELECT * FROM trades WHERE status='open' ORDER BY entry_time DESC"
).fetchall()
return [dict(r) for r in rows]
def close_position(self, trade_id: int, exit_price: float,
exit_time: datetime, fees: float = 0):
"""Close a trade position"""
trade = self.conn.execute(
"SELECT * FROM trades WHERE id=?", (trade_id,)
).fetchone()
if not trade:
return
trade = dict(trade)
if trade['side'] == 'buy':
pnl = (exit_price - trade['entry_price']) * trade['amount'] - fees
pnl_pct = ((exit_price - trade['entry_price']) / trade['entry_price']) * 100
else:
pnl = (trade['entry_price'] - exit_price) * trade['amount'] - fees
pnl_pct = ((trade['entry_price'] - exit_price) / trade['entry_price']) * 100
self.conn.execute(
"""UPDATE trades SET exit_price=?, exit_time=?, pnl=?, pnl_pct=?,
fees=?, status='closed' WHERE id=?""",
(exit_price, exit_time.isoformat(), round(pnl, 4),
round(pnl_pct, 4), fees, trade_id)
)
self.conn.commit()
# --- Strategy performance ---
def record_strategy_result(self, strategy_id: str, params: Dict, metrics: Dict):
"""Record a backtest or live strategy result"""
self.conn.execute(
"""INSERT INTO strategy_performance
(strategy_id, params, backtest_start, backtest_end, total_trades,
win_rate, profit_factor, sharpe_ratio, sortino_ratio, max_drawdown,
total_return, avg_trade_pnl, recorded_at, source)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(strategy_id, json.dumps(params),
metrics.get('backtest_start'), metrics.get('backtest_end'),
metrics.get('total_trades', 0), metrics.get('win_rate', 0),
metrics.get('profit_factor', 0), metrics.get('sharpe_ratio', 0),
metrics.get('sortino_ratio', 0), metrics.get('max_drawdown', 0),
metrics.get('total_return', 0), metrics.get('avg_trade_pnl', 0),
datetime.now().isoformat(), metrics.get('source', 'backtest'))
)
self.conn.commit()
def get_best_strategies(self, metric: str = "sharpe_ratio",
limit: int = 10) -> List[Dict]:
"""Get top performing strategies"""
rows = self.conn.execute(
f"SELECT * FROM strategy_performance ORDER BY {metric} DESC LIMIT ?",
(limit,)
).fetchall()
return [dict(r) for r in rows]
# --- Model checkpoints ---
def save_model_checkpoint(self, model_name: str, epoch: int,
state_dict_bytes: bytes, metrics: Dict = None):
"""Save a model checkpoint"""
self.conn.execute(
"""INSERT OR REPLACE INTO model_checkpoints
(model_name, epoch, state_dict, metrics, saved_at)
VALUES (?, ?, ?, ?, ?)""",
(model_name, epoch, state_dict_bytes,
json.dumps(metrics or {}), datetime.now().isoformat())
)
self.conn.commit()
logger.debug(f"Saved checkpoint: {model_name} epoch {epoch}")
def load_latest_checkpoint(self, model_name: str) -> Optional[Dict]:
"""Load the most recent model checkpoint"""
row = self.conn.execute(
"""SELECT * FROM model_checkpoints
WHERE model_name=? ORDER BY epoch DESC LIMIT 1""",
(model_name,)
).fetchone()
if row:
return dict(row)
return None
# --- Evolution history ---
def record_generation(self, generation: int, population: List[Dict],
best_fitness: float, avg_fitness: float = 0):
"""Record a GA generation"""
self.conn.execute(
"""INSERT INTO evolution_history
(generation, population, best_fitness, avg_fitness, recorded_at)
VALUES (?, ?, ?, ?, ?)""",
(generation, json.dumps(population), best_fitness,
avg_fitness, datetime.now().isoformat())
)
self.conn.commit()
def get_latest_generation(self) -> Optional[Dict]:
"""Load the most recent GA generation"""
row = self.conn.execute(
"SELECT * FROM evolution_history ORDER BY generation DESC LIMIT 1"
).fetchone()
if row:
result = dict(row)
result['population'] = json.loads(result['population'])
return result
return None
# --- System state ---
def save_state(self, key: str, value: str):
"""Save a system state key-value pair"""
self.conn.execute(
"""INSERT OR REPLACE INTO system_state (key, value, updated_at)
VALUES (?, ?, ?)""",
(key, value, datetime.now().isoformat())
)
self.conn.commit()
def load_state(self, key: str) -> Optional[str]:
"""Load a system state value"""
row = self.conn.execute(
"SELECT value FROM system_state WHERE key=?", (key,)
).fetchone()
return row['value'] if row else None
# --- Portfolio snapshots ---
def save_portfolio_snapshot(self, portfolio_value: float, date: str = None):
"""Save end-of-day portfolio snapshot"""
if date is None:
date = datetime.utcnow().strftime("%Y-%m-%d")
try:
self.conn.execute(
"""INSERT OR REPLACE INTO portfolio_snapshots
(date, portfolio_value, recorded_at)
VALUES (?, ?, ?)""",
(date, portfolio_value, datetime.utcnow().isoformat())
)
self.conn.commit()
logger.debug(f"Saved portfolio snapshot: {date} = ${portfolio_value:.2f}")
except Exception as e:
logger.error(f"Error saving portfolio snapshot: {e}")
def get_last_portfolio_snapshot(self, days_ago: int = 1) -> Optional[float]:
"""Get portfolio value from N days ago"""
try:
target_date = (datetime.utcnow() - timedelta(days=days_ago)).strftime("%Y-%m-%d")
row = self.conn.execute(
"""SELECT portfolio_value FROM portfolio_snapshots
WHERE date <= ? ORDER BY date DESC LIMIT 1""",
(target_date,)
).fetchone()
if row:
return float(row[0])
# Fallback to None if no snapshot exists
return None
except Exception as e:
logger.error(f"Error fetching portfolio snapshot: {e}")
return None
def close(self):
"""Close the database connection"""
if self._conn:
self._conn.close()
self._conn = None
src/main_auto.py.backup_20260302_235043
+778
View File
@@ -0,0 +1,778 @@
"""
BIGGFISH Autonomous Self-Learning Trading Bot
24/7 entry point - runs forever, learns continuously, trades autonomously.
Usage:
python -m src.main_auto
python -m src.main_auto --config config/auto_config.json
"""
import json
import sys
import time
import threading
import signal as signal_module
from pathlib import Path
from datetime import datetime, timedelta
from loguru import logger
import numpy as np
# Setup logging
log_path = Path(__file__).parent.parent / "logs"
log_path.mkdir(exist_ok=True)
logger.add(
log_path / "biggfish_auto_{time}.log",
rotation="1 day",
retention="30 days",
level="INFO"
)
from trading.broker import AlpacaBroker
from data.store import DataStore
from data.candle_cache import CandleCache
try:
from trading.oanda_broker import OandaBroker
except ImportError:
OandaBroker = None
from data.features import FeatureEngine
from backtest.engine import BacktestEngine
from backtest.metrics import compute_metrics
from ml.rl_agent import RLAgent
from ml.rl_environment import TradingEnvironment
from ml.genetic import GeneticEvolver, StrategyGenome
from strategies.auto_strategy import genome_to_strategy, evaluate_genome_signal
from trading.executor import TradingExecutor
from core.safety import SafetyManager
from reporting.telegram_reporter import TelegramReporter
from reporting.krystie_bridge import KrystieBridge
class BiggFishAuto:
"""
24/7 Autonomous self-learning trading bot.
Coordinates trading, backtesting, RL training, and GA evolution.
"""
def __init__(self, config_path: str = "config/auto_config.json"):
logger.info("Initializing BIGGFISH Autonomous Trader...")
# Load config
with open(config_path) as f:
self.config = json.load(f)
# Initialize components
self.store = DataStore(self.config['database']['path'])
self.store.initialize()
self.broker = AlpacaBroker(self.config['alpaca'])
self.feature_engine = FeatureEngine()
# OANDA broker for forex (optional)
self.oanda_broker = None
oanda_config = self.config.get('oanda')
if oanda_config and oanda_config.get('api_token') and OandaBroker:
try:
self.oanda_broker = OandaBroker(oanda_config)
logger.info("OANDA forex broker connected")
except Exception as e:
logger.warning(f"OANDA connection failed (forex disabled): {e}")
self.candle_cache = CandleCache(self.store, oanda_broker=self.oanda_broker)
self.backtest_engine = BacktestEngine(
initial_capital=self.config['backtest']['initial_capital'],
commission_rate=self.config['trading'].get('commission_rate', 0.001)
)
self.safety = SafetyManager(self.config['safety'], self.store)
# Executors per broker
self.executor = TradingExecutor(
self.broker, self.store, self.safety, self.config['trading']
)
self.forex_executor = None
if self.oanda_broker:
self.forex_executor = TradingExecutor(
self.oanda_broker, self.store, self.safety, self.config['trading']
)
# RL Agent
env_temp = TradingEnvironment(self.feature_engine,
initial_capital=self.config['trading']['initial_capital'])
self.rl_agent = RLAgent(
state_dim=env_temp.state_dim,
action_dim=TradingEnvironment.NUM_ACTIONS,
config=self.config['rl']
)
# GA Evolver
self.ga_evolver = GeneticEvolver(
self.config['ga'], self.backtest_engine, self.store
)
# Telegram daily reporter
tg_config = self.config.get('telegram', {})
if tg_config.get('enabled') and tg_config.get('bot_token') and tg_config.get('chat_id'):
self.telegram = TelegramReporter(tg_config['bot_token'], tg_config['chat_id'])
logger.info("Telegram reporting enabled")
else:
self.telegram = None
logger.info("Telegram reporting disabled (no config)")
# Krystie bridge (writes status/events to JSON for Krystie to read)
self.krystie = KrystieBridge(self.config['database']['path'].rsplit('/', 1)[0] or 'data')
# State
self.running = False
self.start_time = None
self.cycle_count = 0
self.last_backtest = datetime.min
self.last_rl_train = datetime.min
self.last_ga_evolve = datetime.min
self.last_dashboard = datetime.min
self.last_daily_report = datetime.min
self.recent_trades = [] # Last 10 trades for dashboard
logger.info("BIGGFISH Autonomous Trader initialized")
def start(self):
"""Start the autonomous trading system"""
self.running = True
self.start_time = datetime.utcnow()
# Register shutdown handler
signal_module.signal(signal_module.SIGINT, self._signal_handler)
logger.info("=" * 60)
logger.info(" BIGGFISH AUTONOMOUS TRADER STARTING")
logger.info("=" * 60)
# 1. Load saved state
self._load_state()
self.krystie.log_startup()
# 2. Warm candle cache
self._warm_cache()
# 3. Initial GA population
self.ga_evolver.initialize_population()
if not self.ga_evolver.population:
logger.info("Running initial GA evolution...")
self._run_ga_evolution()
# 4. Initial RL training if no checkpoint
if self.rl_agent.steps == 0:
logger.info("Running initial RL training...")
self._run_rl_training()
# 5. Main loop
logger.info("Entering main loop...")
self._print_dashboard()
while self.running:
try:
cycle_start = datetime.utcnow()
# Trading cycle (run if any market is open)
any_market_open = self.broker.is_market_open()
if self.oanda_broker:
any_market_open = any_market_open or self.oanda_broker.is_market_open()
if any_market_open:
self._trading_cycle()
else:
logger.debug("All markets closed - running learning tasks")
# Periodic tasks (run regardless of market hours)
now = datetime.utcnow()
# Cache update (every 5 min)
cache_interval = self.config['cache']['update_interval_seconds']
if (now - self.last_backtest).total_seconds() > cache_interval:
self._update_cache()
# Backtest (every 30 min)
bt_interval = self.config['backtest']['interval_seconds']
if (now - self.last_backtest).total_seconds() > bt_interval:
self._run_backtest()
self.last_backtest = now
# RL training (every 2 hours)
rl_interval = self.config['rl']['train_interval_hours'] * 3600
if (now - self.last_rl_train).total_seconds() > rl_interval:
self._run_rl_training()
self.last_rl_train = now
# GA evolution (every 6 hours)
ga_interval = self.config['ga']['evolution_interval_hours'] * 3600
if (now - self.last_ga_evolve).total_seconds() > ga_interval:
self._run_ga_evolution()
self.last_ga_evolve = now
# Daily Telegram report (once per day, around market close ~21:00 UTC)
if self.telegram and self._should_send_daily_report(now):
self._send_daily_telegram_report()
self.last_daily_report = now
# Dashboard (every minute)
dash_interval = self.config['reporting']['dashboard_interval_seconds']
if (now - self.last_dashboard).total_seconds() > dash_interval:
self._print_dashboard()
self.last_dashboard = now
# Sleep until next cycle
elapsed = (datetime.utcnow() - cycle_start).total_seconds()
sleep_time = max(1, self.config['trading']['cycle_interval_seconds'] - elapsed)
time.sleep(sleep_time)
except KeyboardInterrupt:
self._shutdown()
break
except Exception as e:
logger.error(f"Main loop error: {e}", exc_info=True)
time.sleep(30)
def _is_forex(self, symbol: str) -> bool:
"""Check if symbol is a forex pair (OANDA format: XXX_YYY)"""
return '_' in symbol and len(symbol) == 7
def _get_broker(self, symbol: str):
"""Get the appropriate broker for a symbol"""
if self._is_forex(symbol) and self.oanda_broker:
return self.oanda_broker
return self.broker
def _get_executor(self, symbol: str):
"""Get the appropriate executor for a symbol"""
if self._is_forex(symbol) and self.forex_executor:
return self.forex_executor
return self.executor
def _trading_cycle(self):
"""Run one trading cycle across all markets"""
self.cycle_count += 1
all_symbols = self._get_tradeable_symbols()
# Check exits first
try:
current_prices = {}
for symbol in all_symbols:
broker = self._get_broker(symbol)
price = broker.get_latest_price(symbol)
if price:
current_prices[symbol] = price
# Check exits per executor
stock_prices = {s: p for s, p in current_prices.items() if not self._is_forex(s)}
forex_prices = {s: p for s, p in current_prices.items() if self._is_forex(s)}
if stock_prices:
closed = self.executor.check_exits(stock_prices)
for trade in closed:
logger.info(f"CLOSED {trade['symbol']}: ${trade['pnl']:+.2f} "
f"({trade['pnl_pct']:+.1f}%) [{trade['exit_reason']}]")
self.recent_trades.append(trade)
self.krystie.log_trade(trade)
if forex_prices and self.forex_executor:
closed = self.forex_executor.check_exits(forex_prices)
for trade in closed:
logger.info(f"CLOSED {trade['symbol']}: ${trade['pnl']:+.2f} "
f"({trade['pnl_pct']:+.1f}%) [{trade['exit_reason']}]")
self.recent_trades.append(trade)
self.krystie.log_trade(trade)
except Exception as e:
logger.error(f"Exit check error: {e}")
# Check safety
if not self.safety.is_trading_allowed():
logger.warning(f"Trading halted: {self.safety.halt_reason}")
return
# Evaluate each symbol
for symbol in all_symbols:
try:
self._trade_symbol(symbol)
except Exception as e:
logger.debug(f"Error trading {symbol}: {e}")
# Keep only last 20 trades
self.recent_trades = self.recent_trades[-20:]
def _get_tradeable_symbols(self) -> list:
"""Get symbols that can be traded right now"""
stock_symbols = self.config['trading'].get('symbols', [])
forex_symbols = self.config['trading'].get('forex_symbols', [])
tradeable = []
# Stocks: only if Alpaca market is open
if self.broker.is_market_open():
tradeable.extend(stock_symbols)
# Forex: if OANDA is connected and forex market is open (24/5)
if self.oanda_broker and self.oanda_broker.is_market_open():
tradeable.extend(forex_symbols)
return tradeable
def _trade_symbol(self, symbol: str):
"""Evaluate and potentially trade a single symbol"""
broker = self._get_broker(symbol)
executor = self._get_executor(symbol)
# Get candle data
df = self.candle_cache.get_cached(
symbol, '1h',
start=datetime.utcnow() - timedelta(days=30)
)
if df is None or len(df) < 60:
return
# Compute features
features_df = self.feature_engine.compute_and_normalize(df)
if features_df is None or len(features_df) < 10:
return
# Get market state
state = self.feature_engine.get_state_vector(features_df, -1)
# Get GA signal
best_genome = self.ga_evolver.get_best_genome()
ga_signal = {'signal': 'hold', 'confidence': 0.0}
strategy_params = {}
if best_genome:
raw_df = self.feature_engine.compute(df)
if raw_df is not None and len(raw_df) > 0:
ga_signal = evaluate_genome_signal(best_genome, raw_df, len(raw_df) - 1)
strategy_params = {
'stop_loss': ga_signal.get('stop_loss'),
'take_profit': ga_signal.get('take_profit'),
'position_pct': ga_signal.get('position_pct', 0.1),
'strategy_id': f"ga_gen{best_genome.generation}",
}
# Augment state with GA signal + portfolio
portfolio_state = executor.get_portfolio_state()
signal_dir = 1.0 if ga_signal['signal'] == 'buy' else (
-1.0 if ga_signal['signal'] == 'sell' else 0.0)
augmented_state = np.concatenate([
state,
[portfolio_state.get('position_ratio', 0),
portfolio_state.get('unrealized_pnl', 0),
portfolio_state.get('time_in_position', 0)],
[signal_dir, ga_signal.get('confidence', 0.0)],
])
# RL agent decides
action = self.rl_agent.select_action(augmented_state, live_mode=True)
if action == 0: # Hold
return
# Get current price
current_price = broker.get_latest_price(symbol)
if not current_price:
return
# Execute
trade = executor.execute_signal(
symbol, action, current_price, strategy_params
)
if trade:
action_name = TradingEnvironment.ACTION_NAMES[action]
logger.info(f"TRADE: {action_name} {symbol} @ ${current_price:.4f}")
self.recent_trades.append(trade)
self.krystie.log_trade(trade)
def _all_symbols(self) -> list:
"""Get all configured symbols (stocks + forex)"""
symbols = list(self.config['trading'].get('symbols', []))
symbols.extend(self.config['trading'].get('forex_symbols', []))
return symbols
def _warm_cache(self):
"""Warm the candle cache"""
logger.info("Warming candle cache...")
symbols = self._all_symbols()
timeframes = self.config['cache'].get('timeframes', ['1h'])
lookback = self.config['cache'].get('warmup_lookback_days', 90)
self.candle_cache.warm_cache(symbols, timeframes, lookback)
logger.info("Cache warmup complete")
def _update_cache(self):
"""Update candle cache incrementally"""
symbols = self._all_symbols()
timeframes = self.config['cache'].get('timeframes', ['1h'])
self.candle_cache.update_cache(symbols, timeframes)
def _run_backtest(self):
"""Run background backtest of current strategy"""
best_genome = self.ga_evolver.get_best_genome()
if not best_genome:
return
strategy_fn = genome_to_strategy(best_genome)
symbols = self._all_symbols()[:3] # Top 3 for speed
for symbol in symbols:
df = self.candle_cache.get_cached(
symbol, '1h',
start=datetime.utcnow() - timedelta(days=30)
)
if df is None or len(df) < 60:
continue
featured_df = self.feature_engine.compute(df)
if featured_df is None or len(featured_df) < 50:
continue
result = self.backtest_engine.run(
strategy_fn, featured_df,
params=best_genome.to_dict(), symbol=symbol
)
if result.metrics.get('total_trades', 0) > 0:
self.store.record_strategy_result(
strategy_id=f"ga_gen{best_genome.generation}",
params=best_genome.to_dict(),
metrics=result.metrics
)
logger.info(f"Backtest {symbol}: Sharpe={result.metrics['sharpe_ratio']:.2f} "
f"WR={result.metrics['win_rate']:.0f}% "
f"Return={result.metrics['total_return']:.1f}%")
def _run_rl_training(self):
"""Run RL batch training"""
logger.info("Starting RL training...")
env = TradingEnvironment(
self.feature_engine,
initial_capital=self.config['trading']['initial_capital']
)
total_metrics = {'avg_loss': 0, 'episodes': 0}
for symbol in self._all_symbols()[:5]:
df = self.candle_cache.get_cached(
symbol, '1h',
start=datetime.utcnow() - timedelta(days=30)
)
if df is None or len(df) < 100:
continue
# Create GA signal function for this data
best_genome = self.ga_evolver.get_best_genome()
ga_fn = None
if best_genome:
featured = self.feature_engine.compute(df)
if featured is not None and len(featured) > 0:
def make_ga_fn(genome, feat_df):
def fn(step):
if step < len(feat_df):
return evaluate_genome_signal(genome, feat_df, step)
return {'signal': 'hold', 'confidence': 0.0}
return fn
ga_fn = make_ga_fn(best_genome, featured)
metrics = self.rl_agent.train_on_episode(env, df, ga_signal_fn=ga_fn)
total_metrics['avg_loss'] += metrics.get('avg_loss', 0)
total_metrics['episodes'] += 1
# Save checkpoint
self.rl_agent.save(self.store)
if total_metrics['episodes'] > 0:
avg = total_metrics['avg_loss'] / total_metrics['episodes']
logger.info(f"RL training complete: {total_metrics['episodes']} episodes, "
f"avg_loss={avg:.6f}, epsilon={self.rl_agent.epsilon:.4f}")
def _run_ga_evolution(self):
"""Run GA evolution cycle"""
logger.info("Starting GA evolution...")
# Prepare candle data for fitness evaluation
candles_data = {}
symbols = self._all_symbols()[:5]
for symbol in symbols:
df = self.candle_cache.get_cached(
symbol, '1h',
start=datetime.utcnow() - timedelta(days=30)
)
if df is not None and len(df) > 60:
featured = self.feature_engine.compute(df)
if featured is not None and len(featured) > 50:
candles_data[symbol] = featured
if not candles_data:
logger.warning("No data available for GA evolution")
return
# Strategy function factory
def strategy_fn_factory(genome):
return genome_to_strategy(genome)
# Run evolution
num_gens = self.config['ga'].get('generations_per_cycle', 10)
best = self.ga_evolver.run_evolution_cycle(
strategy_fn_factory, candles_data, num_generations=num_gens
)
if best:
logger.info(f"GA evolution complete: Gen {self.ga_evolver.generation}, "
f"best fitness={best.fitness:.4f}")
self.krystie.log_ga_milestone(self.ga_evolver.generation, best.fitness)
def _should_send_daily_report(self, now: datetime) -> bool:
"""Check if it's time to send the daily report (once per day after 21:00 UTC)"""
report_hour = self.config.get('telegram', {}).get('daily_report_hour', 21)
if now.hour >= report_hour and (now - self.last_daily_report).total_seconds() > 20 * 3600:
return True
return False
def _send_daily_telegram_report(self):
"""Gather data and send the daily Telegram report"""
try:
portfolio = self.broker.get_portfolio()
positions = self.broker.get_positions()
# Add OANDA positions if available
if self.oanda_broker:
try:
oanda_positions = self.oanda_broker.get_positions()
positions.extend(oanda_positions)
except Exception:
pass
# Get today's trades from DB
today_start = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
today_trades = self.store.get_trades(start=today_start, limit=50)
# Learning stats
rl_stats = self.rl_agent.get_stats()
best_genome = self.ga_evolver.get_best_genome()
learning_stats = {
'rl': rl_stats,
'ga': {
'generation': self.ga_evolver.generation,
'best_fitness': best_genome.fitness if best_genome else 0,
},
}
self.telegram.send_daily_report(
portfolio, positions, today_trades, learning_stats,
self.config['trading']
)
self.krystie.log_daily_report(
equity=portfolio.get('equity', 0),
day_pnl=portfolio.get('day_pnl', 0),
trades_count=len(today_trades),
)
logger.info("Daily Telegram report sent")
except Exception as e:
logger.error(f"Failed to send daily Telegram report: {e}")
def _print_dashboard(self):
"""Print live console dashboard"""
try:
portfolio = self.broker.get_portfolio()
positions = self.broker.get_positions()
except Exception as e:
logger.error(f"Dashboard error: {e}")
return
equity = portfolio['equity']
target = self.config['trading']['target_capital']
initial = self.config['trading']['initial_capital']
progress = (equity / target) * 100
total_pnl = equity - initial
total_pnl_pct = (total_pnl / initial) * 100
uptime = datetime.utcnow() - self.start_time if self.start_time else timedelta()
hours = int(uptime.total_seconds() // 3600)
minutes = int((uptime.total_seconds() % 3600) // 60)
best_genome = self.ga_evolver.get_best_genome()
gen = self.ga_evolver.generation
best_fit = best_genome.fitness if best_genome else 0
rl_stats = self.rl_agent.get_stats()
safety_status = self.safety.get_status()
stock_status = "OPEN" if self.broker.is_market_open() else "CLOSED"
forex_status = ""
if self.oanda_broker:
forex_status = " | FX: " + ("OPEN" if self.oanda_broker.is_market_open() else "CLOSED")
# Build dashboard
lines = []
lines.append("")
lines.append("=" * 64)
lines.append(f" BIGGFISH AUTONOMOUS TRADER | {hours}h {minutes}m | "
f"Gen {gen} | Stocks: {stock_status}{forex_status}")
lines.append("=" * 64)
# Progress bar
bar_width = 30
filled = int(bar_width * min(progress, 100) / 100)
bar = "#" * filled + "-" * (bar_width - filled)
lines.append(f" Portfolio: ${equity:.2f} / ${target} [{bar}] {progress:.1f}%")
lines.append(f" Day P&L: ${portfolio['day_pnl']:+.2f} ({portfolio['day_pnl_pct']:+.1f}%)")
lines.append(f" Total P&L: ${total_pnl:+.2f} ({total_pnl_pct:+.1f}%)")
# Positions
lines.append("-" * 64)
if positions:
lines.append(f" Active Positions ({len(positions)}):")
for p in positions[:5]:
pl_str = f"${p['unrealized_pl']:+.2f} ({p['unrealized_plpc']:+.1f}%)"
lines.append(f" {p['symbol']:6s} {p['qty']:.0f} @ ${p['avg_entry_price']:.2f}"
f" -> ${p['current_price']:.2f} {pl_str}")
else:
lines.append(" No active positions")
# Learning status
lines.append("-" * 64)
lines.append(" Learning Status:")
lines.append(f" RL Agent: epsilon={rl_stats['epsilon']:.3f} | "
f"loss={rl_stats['avg_loss']:.6f} | "
f"{rl_stats['memory_size']:,} experiences")
lines.append(f" GA: gen {gen} | best fitness={best_fit:.4f}")
# Recent trades
if self.recent_trades:
lines.append("-" * 64)
lines.append(" Recent Trades:")
for t in self.recent_trades[-5:]:
side = t.get('side', '?').upper()
symbol = t.get('symbol', '?')
pnl = t.get('pnl')
if pnl is not None:
pnl_str = f" P&L: ${pnl:+.2f}"
else:
pnl_str = ""
price = t.get('entry_price') or t.get('exit_price', 0)
lines.append(f" {side:4s} {symbol:6s} "
f"{t.get('amount', 0):.1f} @ ${price:.2f}{pnl_str}")
# Safety
if not safety_status['trading_allowed']:
lines.append("-" * 64)
lines.append(f" !! TRADING HALTED: {safety_status['halt_reason']}")
# Next events
lines.append("-" * 64)
now = datetime.utcnow()
bt_next = max(0, self.config['backtest']['interval_seconds'] -
(now - self.last_backtest).total_seconds())
rl_next = max(0, self.config['rl']['train_interval_hours'] * 3600 -
(now - self.last_rl_train).total_seconds())
ga_next = max(0, self.config['ga']['evolution_interval_hours'] * 3600 -
(now - self.last_ga_evolve).total_seconds())
lines.append(f" Next: backtest {bt_next/60:.0f}m | "
f"RL train {rl_next/60:.0f}m | "
f"GA evolve {ga_next/3600:.1f}h")
lines.append("=" * 64)
print("\n".join(lines))
# Update Krystie status file
try:
markets = {"stocks": stock_status}
if self.oanda_broker:
markets["forex"] = "OPEN" if self.oanda_broker.is_market_open() else "CLOSED"
learning_stats = {
'rl': rl_stats,
'ga': {'generation': gen, 'best_fitness': best_fit},
}
today_start = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
today_trades = self.store.get_trades(start=today_start, limit=50)
self.krystie.update_status(
portfolio=portfolio,
positions=positions,
learning_stats=learning_stats,
markets=markets,
config=self.config['trading'],
uptime_seconds=uptime.total_seconds(),
today_trades=today_trades,
)
except Exception as e:
logger.debug(f"Krystie status update error: {e}")
def _load_state(self):
"""Load saved state from database"""
# Load RL model
if self.rl_agent.load(self.store):
logger.info("Loaded RL model from checkpoint")
# Load timing state
last_bt = self.store.load_state('last_backtest')
if last_bt:
self.last_backtest = datetime.fromisoformat(last_bt)
last_rl = self.store.load_state('last_rl_train')
if last_rl:
self.last_rl_train = datetime.fromisoformat(last_rl)
last_ga = self.store.load_state('last_ga_evolve')
if last_ga:
self.last_ga_evolve = datetime.fromisoformat(last_ga)
last_report = self.store.load_state('last_daily_report')
if last_report:
self.last_daily_report = datetime.fromisoformat(last_report)
logger.info("State loaded from database")
def _save_state(self):
"""Save state to database for recovery"""
self.rl_agent.save(self.store)
self.store.save_state('last_backtest', self.last_backtest.isoformat())
self.store.save_state('last_rl_train', self.last_rl_train.isoformat())
self.store.save_state('last_ga_evolve', self.last_ga_evolve.isoformat())
self.store.save_state('last_daily_report', self.last_daily_report.isoformat())
self.store.save_state('last_shutdown', datetime.utcnow().isoformat())
logger.info("State saved to database")
def _signal_handler(self, signum, frame):
"""Handle SIGINT for graceful shutdown"""
self._shutdown()
def _shutdown(self):
"""Graceful shutdown"""
logger.info("Shutting down BIGGFISH...")
self.running = False
self.krystie.log_shutdown()
self._save_state()
self.store.close()
logger.info("Shutdown complete. State saved. Resume anytime.")
def main():
import argparse
parser = argparse.ArgumentParser(description="BIGGFISH Autonomous Trader")
parser.add_argument('--config', default='config/auto_config.json',
help='Path to configuration file')
args = parser.parse_args()
config_path = Path(args.config)
if not config_path.exists():
print(f"Config file not found: {config_path}")
print("Copy config/auto_config.json and fill in your Alpaca API keys")
sys.exit(1)
bot = BiggFishAuto(str(config_path))
bot.start()
if __name__ == '__main__':
main()
src/ml/genetic.py.backup_20260304_235851
+476
View File
@@ -0,0 +1,476 @@
"""
Genetic Algorithm for Strategy Parameter Optimization
Evolves a population of StrategyGenome instances to find profitable trading parameters.
Optimized: uses vectorized backtesting and parallel genome evaluation.
"""
import random
import math
import json
from concurrent.futures import ProcessPoolExecutor, as_completed
from dataclasses import dataclass, field, asdict
from typing import Dict, List, Optional, Tuple
from loguru import logger
# Gene ranges: (min, max, is_int)
GENE_RANGES = {
'fast_ma_period': (5, 50, True),
'slow_ma_period': (20, 200, True),
'rsi_period': (7, 28, True),
'rsi_overbought': (60, 85, False),
'rsi_oversold': (15, 40, False),
'bb_period': (10, 30, True),
'bb_std': (1.5, 3.0, False),
'atr_period': (7, 21, True),
'macd_fast': (8, 16, True),
'macd_slow': (20, 32, True),
'macd_signal': (7, 12, True),
'volume_surge_threshold': (1.2, 3.0, False),
'stop_loss_atr_mult': (1.0, 2.0, False),
'take_profit_atr_mult': (2.5, 5.0, False),
'max_position_pct': (0.05, 0.30, False),
'min_hold_candles': (1, 24, True),
'max_hold_candles': (12, 168, True),
}
@dataclass
class StrategyGenome:
"""A genome encoding all tunable strategy parameters"""
# Indicator periods
fast_ma_period: int = 10
slow_ma_period: int = 50
rsi_period: int = 14
rsi_overbought: float = 70.0
rsi_oversold: float = 30.0
bb_period: int = 20
bb_std: float = 2.0
atr_period: int = 14
macd_fast: int = 12
macd_slow: int = 26
macd_signal: int = 9
# Entry thresholds
volume_surge_threshold: float = 1.5
# Risk management
stop_loss_atr_mult: float = 1.5
take_profit_atr_mult: float = 3.5
max_position_pct: float = 0.20
# Timing
min_hold_candles: int = 2
max_hold_candles: int = 48
# Fitness (set after evaluation)
fitness: float = 0.0
generation: int = 0
def to_dict(self) -> Dict:
return asdict(self)
@classmethod
def from_dict(cls, d: Dict) -> 'StrategyGenome':
valid_fields = {f.name for f in cls.__dataclass_fields__.values()}
filtered = {k: v for k, v in d.items() if k in valid_fields}
return cls(**filtered)
@classmethod
def random(cls, generation: int = 0) -> 'StrategyGenome':
"""Create a random genome within valid parameter ranges"""
genes = {}
for gene_name, (lo, hi, is_int) in GENE_RANGES.items():
if is_int:
genes[gene_name] = random.randint(int(lo), int(hi))
else:
genes[gene_name] = round(random.uniform(lo, hi), 4)
# Constraint: slow_ma > fast_ma
if genes['slow_ma_period'] <= genes['fast_ma_period']:
genes['slow_ma_period'] = genes['fast_ma_period'] + random.randint(10, 50)
# Constraint: macd_slow > macd_fast
if genes['macd_slow'] <= genes['macd_fast']:
genes['macd_slow'] = genes['macd_fast'] + random.randint(8, 16)
# Constraint: take_profit must be at least 1.5x stop_loss (enforces min 1.5:1 R/R)
if genes['take_profit_atr_mult'] < genes['stop_loss_atr_mult'] * 1.5:
genes['take_profit_atr_mult'] = genes['stop_loss_atr_mult'] * random.uniform(1.5, 2.5)
# Constraint: max_hold > min_hold
if genes['max_hold_candles'] <= genes['min_hold_candles']:
genes['max_hold_candles'] = genes['min_hold_candles'] + random.randint(10, 50)
genes['generation'] = generation
return cls(**genes)
def _evaluate_genome_worker(genome_dict: Dict, candles_dict: Dict[str, Dict],
initial_capital: float, commission_rate: float) -> float:
"""
Worker function for parallel genome evaluation.
Runs in a separate process, so must be a top-level function.
Uses vectorized signal generation + fast backtest.
"""
from strategies.auto_strategy import genome_to_signals
from backtest.engine import BacktestEngine
import pandas as pd
genome = StrategyGenome.from_dict(genome_dict)
engine = BacktestEngine(initial_capital=initial_capital, commission_rate=commission_rate)
fitness_scores = []
for symbol, candle_data in candles_dict.items():
df = pd.DataFrame(candle_data)
if len(df) < 60:
continue
try:
signals = genome_to_signals(genome, df)
result = engine.run_fast(signals, df, symbol=symbol)
except Exception:
continue
m = result.metrics
sharpe = max(m.get('sharpe_ratio', 0), 0)
max_dd = m.get('max_drawdown', 0)
total_trades = m.get('total_trades', 0)
dd_penalty = max(1 - max_dd / 100, 0)
trade_bonus = math.sqrt(max(total_trades, 0))
if total_trades < 3:
trade_bonus *= 0.5
score = sharpe * dd_penalty * trade_bonus
fitness_scores.append(score)
if not fitness_scores:
return 0.0
return sum(fitness_scores) / len(fitness_scores)
class GeneticEvolver:
"""
Genetic algorithm for optimizing strategy parameters.
Evolves a population of StrategyGenome instances.
"""
def __init__(self, config: Dict, backtest_engine, store=None):
self.population_size = config.get('population_size', 50)
self.elite_count = config.get('elite_count', 5)
self.mutation_rate = config.get('mutation_rate', 0.15)
self.mutation_strength = config.get('mutation_strength', 0.2)
self.crossover_rate = config.get('crossover_rate', 0.7)
self.tournament_size = config.get('tournament_size', 5)
self.parallel_workers = config.get('parallel_workers', 4)
self.backtest_engine = backtest_engine
self.store = store
self.population: List[StrategyGenome] = []
self.generation = 0
self.best_ever: Optional[StrategyGenome] = None
def initialize_population(self):
"""
Create initial population.
If evolution history exists in DB, load the latest generation.
Otherwise, create random genomes.
"""
if self.store:
latest = self.store.get_latest_generation()
if latest:
self.generation = latest['generation']
self.population = [
StrategyGenome.from_dict(g) for g in latest['population']
]
if self.population:
self.best_ever = max(self.population, key=lambda g: g.fitness)
logger.info(f"Loaded GA population from generation {self.generation} "
f"({len(self.population)} genomes)")
return
# Create random population
self.population = [
StrategyGenome.random(generation=0)
for _ in range(self.population_size)
]
self.generation = 0
logger.info(f"Created random population of {self.population_size} genomes")
def evaluate_fitness(self, genome: StrategyGenome, strategy_fn_factory,
candles_data: Dict[str, 'pd.DataFrame']) -> float:
"""
Evaluate a genome using vectorized fast backtest.
Falls back to callback-based if signals not available.
"""
from strategies.auto_strategy import genome_to_signals
fitness_scores = []
for symbol, df in candles_data.items():
if df is None or len(df) < 60:
continue
try:
signals = genome_to_signals(genome, df)
result = self.backtest_engine.run_fast(signals, df, symbol=symbol)
except Exception:
# Fallback to callback-based
strategy_fn = strategy_fn_factory(genome)
result = self.backtest_engine.run(
strategy_fn, df, params=genome.to_dict(), symbol=symbol
)
m = result.metrics
sharpe = max(m.get('sharpe_ratio', 0), 0)
max_dd = m.get('max_drawdown', 0)
total_trades = m.get('total_trades', 0)
dd_penalty = max(1 - max_dd / 100, 0)
trade_bonus = math.sqrt(max(total_trades, 0))
if total_trades < 3:
trade_bonus *= 0.5
score = sharpe * dd_penalty * trade_bonus
fitness_scores.append(score)
if not fitness_scores:
return 0.0
return sum(fitness_scores) / len(fitness_scores)
def evaluate_population(self, strategy_fn_factory,
candles_data: Dict[str, 'pd.DataFrame']):
"""Evaluate all genomes - uses parallel workers if available."""
unevaluated = [(i, g) for i, g in enumerate(self.population) if g.fitness == 0.0]
if not unevaluated:
return
# Prepare serializable candle data for parallel workers
candles_dict = {}
for symbol, df in candles_data.items():
if df is not None and len(df) >= 60:
candles_dict[symbol] = {
'open': df['open'].values.tolist(),
'high': df['high'].values.tolist(),
'low': df['low'].values.tolist(),
'close': df['close'].values.tolist(),
'volume': df['volume'].values.tolist(),
}
if not candles_dict:
return
initial_capital = self.backtest_engine.initial_capital
commission_rate = self.backtest_engine.commission_rate
# Try parallel evaluation
if self.parallel_workers > 1 and len(unevaluated) > 4:
try:
self._evaluate_parallel(
unevaluated, candles_dict, initial_capital,
commission_rate, strategy_fn_factory, candles_data
)
return
except Exception as e:
logger.debug(f"Parallel evaluation failed, falling back to sequential: {e}")
# Sequential fallback (still uses vectorized fast path)
for idx, (i, genome) in enumerate(unevaluated):
genome.fitness = self.evaluate_fitness(
genome, strategy_fn_factory, candles_data
)
if (idx + 1) % 10 == 0:
logger.debug(f"Evaluated {idx+1}/{len(unevaluated)} genomes")
def _evaluate_parallel(self, unevaluated, candles_dict, initial_capital,
commission_rate, strategy_fn_factory, candles_data):
"""Evaluate genomes in parallel using ProcessPoolExecutor."""
genome_dicts = [(i, g.to_dict()) for i, g in unevaluated]
with ProcessPoolExecutor(max_workers=self.parallel_workers) as executor:
futures = {}
for i, gd in genome_dicts:
fut = executor.submit(
_evaluate_genome_worker, gd, candles_dict,
initial_capital, commission_rate
)
futures[fut] = i
done_count = 0
for future in as_completed(futures):
pop_idx = futures[future]
try:
fitness = future.result(timeout=30)
self.population[pop_idx].fitness = fitness
except Exception:
# Fallback for this genome
self.population[pop_idx].fitness = self.evaluate_fitness(
self.population[pop_idx], strategy_fn_factory, candles_data
)
done_count += 1
if done_count % 10 == 0:
logger.debug(f"Evaluated {done_count}/{len(unevaluated)} genomes (parallel)")
def select_parent(self) -> StrategyGenome:
"""Tournament selection"""
tournament = random.sample(
self.population,
min(self.tournament_size, len(self.population))
)
return max(tournament, key=lambda g: g.fitness)
def crossover(self, parent1: StrategyGenome,
parent2: StrategyGenome) -> StrategyGenome:
"""Uniform crossover: for each gene, randomly pick from parent1 or parent2"""
child_genes = {}
p1 = parent1.to_dict()
p2 = parent2.to_dict()
for gene_name in GENE_RANGES:
child_genes[gene_name] = p1[gene_name] if random.random() < 0.5 else p2[gene_name]
# Repair constraints
if child_genes['slow_ma_period'] <= child_genes['fast_ma_period']:
child_genes['slow_ma_period'] = child_genes['fast_ma_period'] + 10
if child_genes['macd_slow'] <= child_genes['macd_fast']:
child_genes['macd_slow'] = child_genes['macd_fast'] + 8
if child_genes['take_profit_atr_mult'] <= child_genes['stop_loss_atr_mult']:
child_genes['take_profit_atr_mult'] = child_genes['stop_loss_atr_mult'] + 0.5
if child_genes['max_hold_candles'] <= child_genes['min_hold_candles']:
child_genes['max_hold_candles'] = child_genes['min_hold_candles'] + 10
child_genes['fitness'] = 0.0
child_genes['generation'] = self.generation + 1
return StrategyGenome(**child_genes)
def mutate(self, genome: StrategyGenome) -> StrategyGenome:
"""Gaussian mutation on each gene with probability mutation_rate"""
genes = genome.to_dict()
for gene_name, (lo, hi, is_int) in GENE_RANGES.items():
if random.random() < self.mutation_rate:
gene_range = hi - lo
delta = random.gauss(0, gene_range * self.mutation_strength)
new_val = genes[gene_name] + delta
new_val = max(lo, min(hi, new_val))
if is_int:
new_val = round(new_val)
else:
new_val = round(new_val, 4)
genes[gene_name] = new_val
# Repair constraints after mutation
if genes['slow_ma_period'] <= genes['fast_ma_period']:
genes['slow_ma_period'] = genes['fast_ma_period'] + 10
if genes['macd_slow'] <= genes['macd_fast']:
genes['macd_slow'] = genes['macd_fast'] + 8
if genes['take_profit_atr_mult'] <= genes['stop_loss_atr_mult']:
genes['take_profit_atr_mult'] = genes['stop_loss_atr_mult'] + 0.5
if genes['max_hold_candles'] <= genes['min_hold_candles']:
genes['max_hold_candles'] = genes['min_hold_candles'] + 10
genes['fitness'] = 0.0
genes['generation'] = self.generation + 1
return StrategyGenome.from_dict(genes)
def evolve_generation(self, strategy_fn_factory,
candles_data: Dict[str, 'pd.DataFrame']) -> Dict:
"""
Run one generation of evolution:
1. Evaluate fitness of all genomes
2. Sort by fitness
3. Keep elite_count best unchanged
4. Fill remaining via tournament selection + crossover + mutation
5. Save to database
"""
# Evaluate
self.evaluate_population(strategy_fn_factory, candles_data)
# Sort by fitness
self.population.sort(key=lambda g: g.fitness, reverse=True)
best = self.population[0]
avg_fitness = sum(g.fitness for g in self.population) / len(self.population)
if self.best_ever is None or best.fitness > self.best_ever.fitness:
self.best_ever = StrategyGenome.from_dict(best.to_dict())
self.best_ever.fitness = best.fitness
# Elitism: keep top N unchanged
new_population = [
StrategyGenome.from_dict(g.to_dict())
for g in self.population[:self.elite_count]
]
# Preserve their fitness
for i in range(min(self.elite_count, len(self.population))):
new_population[i].fitness = self.population[i].fitness
# Fill the rest
while len(new_population) < self.population_size:
parent1 = self.select_parent()
parent2 = self.select_parent()
if random.random() < self.crossover_rate:
child = self.crossover(parent1, parent2)
else:
child = StrategyGenome.from_dict(parent1.to_dict())
child.fitness = 0.0
child = self.mutate(child)
new_population.append(child)
self.population = new_population
self.generation += 1
# Save to database
if self.store:
self.store.record_generation(
self.generation,
[g.to_dict() for g in self.population],
best.fitness,
avg_fitness
)
stats = {
'generation': self.generation,
'best_fitness': round(best.fitness, 4),
'avg_fitness': round(avg_fitness, 4),
'best_genome': best.to_dict(),
}
logger.info(f"GA Gen {self.generation}: best={best.fitness:.4f} avg={avg_fitness:.4f}")
return stats
def run_evolution_cycle(self, strategy_fn_factory,
candles_data: Dict[str, 'pd.DataFrame'],
num_generations: int = 10) -> Optional[StrategyGenome]:
"""
Run multiple generations.
Returns the best genome found.
"""
for _ in range(num_generations):
self.evolve_generation(strategy_fn_factory, candles_data)
return self.get_best_genome()
def get_best_genome(self) -> Optional[StrategyGenome]:
"""Return the highest-fitness genome"""
if self.best_ever:
return self.best_ever
if self.population:
return max(self.population, key=lambda g: g.fitness)
return None
src/trading/oanda_broker.py.backup
+321
View File
@@ -0,0 +1,321 @@
"""
OANDA broker interface for forex paper trading.
Uses OANDA v20 REST API - no extra dependency, just requests.
"""
import requests
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from loguru import logger
class OandaBroker:
"""OANDA practice (paper) trading broker interface"""
PRACTICE_URL = "https://api-fxpractice.oanda.com"
LIVE_URL = "https://api-fxtrade.oanda.com"
# Forex pairs trade 24/5 (Sun 5PM ET to Fri 5PM ET)
# Map our timeframes to OANDA granularity
TF_MAP = {
'1m': 'M1', '5m': 'M5', '15m': 'M15',
'1h': 'H1', '4h': 'H4', '1d': 'D',
}
def __init__(self, config: Dict):
self.config = config
self.api_token = config['api_token']
self.account_id = config['account_id']
self.base_url = self.PRACTICE_URL if config.get('practice', True) else self.LIVE_URL
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {self.api_token}',
'Content-Type': 'application/json',
})
# Verify connection
try:
acct = self._get(f'/v3/accounts/{self.account_id}/summary')
balance = acct['account']['balance']
currency = acct['account']['currency']
logger.info(f"OANDA connected (PRACTICE) - Balance: {currency} {balance}")
except Exception as e:
logger.error(f"OANDA connection failed: {e}")
raise
def _get(self, path: str, params: Dict = None) -> Dict:
resp = self.session.get(f'{self.base_url}{path}', params=params, timeout=15)
resp.raise_for_status()
return resp.json()
def _post(self, path: str, data: Dict) -> Dict:
resp = self.session.post(f'{self.base_url}{path}', json=data, timeout=15)
resp.raise_for_status()
return resp.json()
def _put(self, path: str, data: Dict) -> Dict:
resp = self.session.put(f'{self.base_url}{path}', json=data, timeout=15)
resp.raise_for_status()
return resp.json()
# --- Account / Portfolio ---
def get_account(self) -> Dict:
data = self._get(f'/v3/accounts/{self.account_id}/summary')
return data['account']
def get_portfolio(self) -> Dict:
acct = self.get_account()
nav = float(acct['NAV'])
balance = float(acct['balance'])
unrealized_pl = float(acct['unrealizedPL'])
pl = float(acct['pl'])
return {
'equity': nav,
'cash': balance,
'buying_power': float(acct.get('marginAvailable', balance)),
'portfolio_value': nav,
'long_market_value': nav - balance,
'day_pnl': unrealized_pl,
'day_pnl_pct': (unrealized_pl / balance * 100) if balance > 0 else 0,
}
def get_positions(self) -> List[Dict]:
data = self._get(f'/v3/accounts/{self.account_id}/openPositions')
positions = []
for p in data.get('positions', []):
# OANDA separates long/short
long_units = int(p['long']['units']) if p['long']['units'] != '0' else 0
short_units = abs(int(p['short']['units'])) if p['short']['units'] != '0' else 0
if long_units > 0:
unrealized = float(p['long']['unrealizedPL'])
avg_price = float(p['long']['averagePrice'])
# Estimate current price from avg + pnl
current_price = avg_price + (unrealized / long_units) if long_units else avg_price
positions.append({
'symbol': p['instrument'],
'qty': long_units,
'market_value': long_units * current_price,
'cost_basis': long_units * avg_price,
'unrealized_pl': unrealized,
'unrealized_plpc': (unrealized / (long_units * avg_price) * 100)
if avg_price > 0 else 0,
'current_price': current_price,
'avg_entry_price': avg_price,
})
if short_units > 0:
unrealized = float(p['short']['unrealizedPL'])
avg_price = float(p['short']['averagePrice'])
current_price = avg_price - (unrealized / short_units) if short_units else avg_price
positions.append({
'symbol': p['instrument'],
'qty': -short_units,
'market_value': short_units * current_price,
'cost_basis': short_units * avg_price,
'unrealized_pl': unrealized,
'unrealized_plpc': (unrealized / (short_units * avg_price) * 100)
if avg_price > 0 else 0,
'current_price': current_price,
'avg_entry_price': avg_price,
})
return positions
def is_market_open(self) -> bool:
"""Forex is open 24/5 - closed Saturday and most of Sunday"""
now = datetime.utcnow()
# Closed: Friday 22:00 UTC to Sunday 22:00 UTC (roughly)
if now.weekday() == 5: # Saturday
return False
if now.weekday() == 6 and now.hour < 22: # Sunday before 22:00
return False
if now.weekday() == 4 and now.hour >= 22: # Friday after 22:00
return False
return True
def get_market_hours(self) -> Dict:
return {
'is_open': self.is_market_open(),
'next_open': None,
'next_close': None,
}
# --- Orders ---
def place_market_order(self, symbol: str, qty, side: str) -> Dict:
"""Place a market order. qty is in units (not lots)."""
units = int(qty) if side == 'buy' else -int(qty)
data = {
'order': {
'type': 'MARKET',
'instrument': symbol,
'units': str(units),
'timeInForce': 'FOK',
}
}
try:
result = self._post(f'/v3/accounts/{self.account_id}/orders', data)
fill = result.get('orderFillTransaction', {})
order_id = fill.get('id', result.get('orderCreateTransaction', {}).get('id', ''))
logger.info(f"OANDA {side.upper()} {abs(units)} {symbol}")
return {
'id': order_id,
'symbol': symbol,
'qty': abs(units),
'side': side,
'type': 'market',
'status': 'filled' if fill else 'pending',
'fill_price': float(fill.get('price', 0)) if fill else 0,
}
except Exception as e:
logger.error(f"OANDA order error: {e}")
raise
def place_limit_order(self, symbol: str, qty, side: str, limit_price: float) -> Dict:
units = int(qty) if side == 'buy' else -int(qty)
data = {
'order': {
'type': 'LIMIT',
'instrument': symbol,
'units': str(units),
'price': f'{limit_price:.5f}',
'timeInForce': 'GTC',
}
}
result = self._post(f'/v3/accounts/{self.account_id}/orders', data)
order = result.get('orderCreateTransaction', {})
return {
'id': order.get('id', ''),
'symbol': symbol,
'qty': abs(units),
'side': side,
'type': 'limit',
'limit_price': limit_price,
'status': 'pending',
}
def cancel_order(self, order_id: str) -> bool:
try:
self._put(f'/v3/accounts/{self.account_id}/orders/{order_id}/cancel', {})
return True
except Exception:
return False
def get_orders(self, status: str = "open") -> List[Dict]:
state = 'PENDING' if status == 'open' else 'ALL'
data = self._get(f'/v3/accounts/{self.account_id}/orders', {'state': state})
return [
{
'id': o['id'],
'symbol': o.get('instrument', ''),
'qty': abs(int(o.get('units', 0))),
'side': 'buy' if int(o.get('units', 0)) > 0 else 'sell',
'type': o.get('type', '').lower(),
'status': o.get('state', '').lower(),
'created_at': o.get('createTime', ''),
}
for o in data.get('orders', [])
]
# --- Market Data ---
def get_bars(self, symbol: str, timeframe: str = "1d", limit: int = 100) -> List[Dict]:
gran = self.TF_MAP.get(timeframe, 'H1')
params = {'granularity': gran, 'count': min(limit, 5000)}
try:
data = self._get(f'/v3/instruments/{symbol}/candles', params)
return self._parse_candles(data.get('candles', []))
except Exception as e:
logger.error(f"OANDA bars error for {symbol}: {e}")
return []
def fetch_bars_range(self, symbol: str, timeframe: str = "1h",
start=None, end=None) -> List[Dict]:
gran = self.TF_MAP.get(timeframe, 'H1')
params = {'granularity': gran, 'price': 'M'}
if start:
if isinstance(start, datetime):
params['from'] = start.strftime('%Y-%m-%dT%H:%M:%SZ')
else:
params['from'] = str(start)
if end:
if isinstance(end, datetime):
params['to'] = end.strftime('%Y-%m-%dT%H:%M:%SZ')
else:
params['to'] = str(end)
if 'from' not in params:
params['count'] = 500
try:
data = self._get(f'/v3/instruments/{symbol}/candles', params)
return self._parse_candles(data.get('candles', []))
except Exception as e:
logger.error(f"OANDA bars range error for {symbol}: {e}")
return []
def get_latest_price(self, symbol: str) -> Optional[float]:
try:
data = self._get(f'/v3/instruments/{symbol}/candles',
{'granularity': 'M1', 'count': 1, 'price': 'M'})
candles = data.get('candles', [])
if candles:
return float(candles[-1]['mid']['c'])
return None
except Exception as e:
logger.error(f"OANDA price error for {symbol}: {e}")
return None
def close_position(self, symbol: str) -> bool:
"""Close entire position for an instrument"""
try:
# Close long
try:
self._put(f'/v3/accounts/{self.account_id}/positions/{symbol}/close',
{'longUnits': 'ALL'})
except Exception:
pass
# Close short
try:
self._put(f'/v3/accounts/{self.account_id}/positions/{symbol}/close',
{'shortUnits': 'ALL'})
except Exception:
pass
logger.info(f"Closed OANDA position: {symbol}")
return True
except Exception as e:
logger.error(f"Error closing OANDA position {symbol}: {e}")
return False
# --- Helpers ---
def _parse_candles(self, candles: List[Dict]) -> List[Dict]:
"""Convert OANDA candles to our standard format"""
result = []
for c in candles:
if not c.get('complete', True) and len(candles) > 1:
continue # Skip incomplete candles unless it's the only one
mid = c.get('mid', {})
ts = c.get('time', '')
try:
dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
ts_ms = int(dt.timestamp() * 1000)
except (ValueError, AttributeError):
continue
result.append({
'timestamp': ts_ms,
'open': float(mid.get('o', 0)),
'high': float(mid.get('h', 0)),
'low': float(mid.get('l', 0)),
'close': float(mid.get('c', 0)),
'volume': int(c.get('volume', 0)),
})
return result
def get_tradeable_instruments(self) -> List[str]:
"""Get list of available forex instruments"""
data = self._get(f'/v3/accounts/{self.account_id}/instruments')
return [i['name'] for i in data.get('instruments', [])
if i.get('type') == 'CURRENCY']