diff --git a/.claude/settings.local.json b/.claude/settings.local.json index 5c9655a..7280c39 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -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:*)" ] } } diff --git a/.gitignore b/.gitignore index 0e8ceb5..74a13f7 100644 --- a/.gitignore +++ b/.gitignore @@ -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/ diff --git a/src/data/__init__.py b/src/data/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/data/candle_cache.py b/src/data/candle_cache.py new file mode 100644 index 0000000..98997b9 --- /dev/null +++ b/src/data/candle_cache.py @@ -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) diff --git a/src/data/features.py b/src/data/features.py new file mode 100644 index 0000000..c0fb050 --- /dev/null +++ b/src/data/features.py @@ -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) diff --git a/src/data/krystie-events.json b/src/data/krystie-events.json new file mode 100644 index 0000000..e540d5e --- /dev/null +++ b/src/data/krystie-events.json @@ -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 + } + } + ] +} \ No newline at end of file diff --git a/src/data/krystie-status.json b/src/data/krystie-status.json new file mode 100644 index 0000000..0025a7a --- /dev/null +++ b/src/data/krystie-status.json @@ -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 + } +} \ No newline at end of file diff --git a/src/data/store.py b/src/data/store.py new file mode 100644 index 0000000..6302580 --- /dev/null +++ b/src/data/store.py @@ -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 diff --git a/src/main_auto.py.backup_20260302_235043 b/src/main_auto.py.backup_20260302_235043 new file mode 100644 index 0000000..5fdc09b --- /dev/null +++ b/src/main_auto.py.backup_20260302_235043 @@ -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() diff --git a/src/ml/genetic.py.backup_20260304_235851 b/src/ml/genetic.py.backup_20260304_235851 new file mode 100644 index 0000000..d85c00f --- /dev/null +++ b/src/ml/genetic.py.backup_20260304_235851 @@ -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 diff --git a/src/trading/oanda_broker.py.backup b/src/trading/oanda_broker.py.backup new file mode 100644 index 0000000..95348b4 --- /dev/null +++ b/src/trading/oanda_broker.py.backup @@ -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']