From 8e676d25beefe52c9788c1855e02b50c5b30b3a7 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 08:13:52 +0000 Subject: [PATCH] ClaudeTrader: implement the RL strategy (tabular Q-learning) Replace the placeholder RLStrategy (always "hold") with a working agent under strategies/rl/: - features.py: causal discrete market states (SuperTrend direction, RSI zone, ATR-relative momentum, volatility regime); prefixes of a series always produce identical states. - environment.py: target-position actions (flat/long/short), log-return reward minus transaction costs, matching the backtester's semantics. - agent.py: tabular Q-learning with Q-margin confidence and atomic JSON save/load. - trainer.py: RLConfig (validated variables: learning, exploration, costs, split, persistence, features) and a seeded train_agent(). RLStrategy trains on the first train_fraction of the data and trades only the remaining bars in backtests, loads saved models only when their feature settings match, and maps targets to buy/sell/exit using the position the backtester now passes in. Supporting changes: - Backtester asks the strategy once per bar, passes the current position, and supports an "exit" action. SuperTrend results are unchanged. - Strategies now receive their strategies.parameters. section from the engine config (previously always empty), and set_parameters refreshes derived attributes atomically. - calculate_supertrend moved to utils/indicators for reuse. - default_config.yaml rl_agent section documents every RL variable. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_015rhcFXLJHUCwo4mn74RAHy --- ClaudeTrader/README.md | 64 ++++- ClaudeTrader/configs/default_config.yaml | 28 +- ClaudeTrader/strategies/__init__.py | 311 ++++++++++++++++++---- ClaudeTrader/strategies/rl/__init__.py | 31 +++ ClaudeTrader/strategies/rl/agent.py | 150 +++++++++++ ClaudeTrader/strategies/rl/environment.py | 120 +++++++++ ClaudeTrader/strategies/rl/features.py | 138 ++++++++++ ClaudeTrader/strategies/rl/trainer.py | 193 ++++++++++++++ ClaudeTrader/tests/test_rl.py | 287 ++++++++++++++++++++ ClaudeTrader/utils/backtest.py | 150 +++++------ ClaudeTrader/utils/indicators.py | 55 ++++ 11 files changed, 1390 insertions(+), 137 deletions(-) create mode 100644 ClaudeTrader/strategies/rl/__init__.py create mode 100644 ClaudeTrader/strategies/rl/agent.py create mode 100644 ClaudeTrader/strategies/rl/environment.py create mode 100644 ClaudeTrader/strategies/rl/features.py create mode 100644 ClaudeTrader/strategies/rl/trainer.py create mode 100644 ClaudeTrader/tests/test_rl.py diff --git a/ClaudeTrader/README.md b/ClaudeTrader/README.md index d5b5d60..46e477d 100644 --- a/ClaudeTrader/README.md +++ b/ClaudeTrader/README.md @@ -13,6 +13,7 @@ It is a decision-support and research tool. It does **not** place orders. | --- | --- | --- | | Claude LLM calls | ✅ Real Anthropic API calls when `ANTHROPIC_API_KEY` is set; canned mock responses otherwise | `models/llm_interface.py` | | SuperTrend strategy | ✅ Implemented (signals on trend flips) | `strategies/__init__.py` | +| RL strategy | ✅ Tabular Q-learning agent with causal state features, in-sample training and out-of-sample backtests, JSON save/load | `strategies/rl/`, `strategies/__init__.py` | | Event-driven backtester | ✅ Implemented: costs, ATR stops/targets, trailing stops, long/short | `utils/backtest.py` | | Risk utilities | ✅ Implemented | `utils/risk.py` | | Performance metrics | ✅ Implemented | `utils/performance.py` | @@ -22,7 +23,7 @@ It is a decision-support and research tool. It does **not** place orders. | Static web dashboard | ✅ Live public prices/news in the browser | `frontend/index.html` | | Market data | ⚠️ **Mock** (synthetic random-walk OHLCV). The ccxt call is stubbed. | `utils/data_fetcher.py` | | RAG / knowledge base | ⚠️ **In-memory mock**: a small built-in document set with mock embeddings; no vector DB | `models/rag_engine.py` | -| Multi-factor and RL strategies | ⚠️ **Placeholders**: factor scores are constants and the RL agent always holds | `strategies/__init__.py` | +| Multi-factor strategy | ⚠️ **Placeholder**: factor scores are constants | `strategies/__init__.py` | | Hybrid strategy | ⚠️ Confidence-weighted vote of SuperTrend + multi-factor, so it inherits the placeholder | `strategies/__init__.py` | | OpenAI / local LLM providers | ⚠️ Mock only | `models/llm_interface.py` | | `generate_report` | ⚠️ Placeholder | `core/engine.py` | @@ -39,7 +40,9 @@ ClaudeTrader/ ├── models/ │ ├── llm_interface.py # Anthropic Claude client (+ mock fallback) │ └── rag_engine.py # In-memory retrieval over built-in trading docs (mock embeddings) -├── strategies/__init__.py # SuperTrend, MultiFactor, RL (placeholder), Hybrid + registry +├── strategies/ +│ ├── __init__.py # SuperTrend, RL, MultiFactor (placeholder), Hybrid + registry +│ └── rl/ # RL features, environment, Q-learning agent, trainer ├── utils/ │ ├── backtest.py # Event-driven backtester │ ├── risk.py # Exit levels, position sizing, trailing stops, R-multiples @@ -136,8 +139,10 @@ including the current bar. Entries fill at that bar's close, with slippage. - **Exits.** It checks ATR stop-loss and take-profit levels intrabar. If both are hit in the same bar, it assumes the stop hit first. If the bar opens beyond a level (a gap), the order fills at the open. -- **Signals.** `buy` / `sell` open a position. The opposite signal closes it, and the entry logic - can then reverse on the same bar. `hold` leaves an open position alone. +- **Signals.** The strategy is asked once per bar, at the close. Its window includes `position` + (1 long, -1 short, 0 flat). `buy` / `sell` open a position. The opposite signal closes it and + reverses on the same bar. `exit` closes without reversing, and `hold` leaves an open position + alone. - **Trailing stops** (optional) ratchet after each bar closes, so a bar never tightens its own stop. - **Sizing** is fixed-fractional (`risk_per_trade` of equity lost at the stop), capped at available cash (no leverage). @@ -165,6 +170,55 @@ print(result["performance"], result["num_trades"]) Each strategy also has `backtest(symbol, period)`, which fetches data and runs the engine. +## RL strategy (`strategies/rl/`) + +`RLStrategy` (registry name `rl_agent`) is a tabular Q-learning agent. It needs only numpy. + +- **State.** One of 36 market states, built from SuperTrend direction, RSI zone, N-bar momentum + relative to ATR, and whether ATR% is above its rolling median. The agent combines this with its + current position. Each bar's state uses only bars up to that point, so prefixes of a series give + identical states. +- **Actions** are target positions: flat, long or short (short only if `allow_short`). They map to + `buy`, `sell`, and `exit` (or `hold` when already flat). +- **Reward** is the log return of the position held into the next bar, minus `transaction_cost` for + each unit of position change. +- **Training** is epsilon-greedy with linear decay. It is reproducible for a given `seed`. +- **Confidence** is the Q-value margin between the best and second-best action. A state the agent + never saw in training yields `hold` with confidence 0. +- **Persistence.** `save()` writes JSON to `model_path`, and so does training when `save_model` is + set. A saved model is loaded at startup when `load_model` is set and its feature settings match + the current config; otherwise it is ignored. + +`backtest()` trains on the first `train_fraction` of the data and trades only the remaining bars, +so results are out-of-sample. An agent trained earlier in memory is retrained for every backtest. +An agent loaded from disk is used as-is. The backtester's ATR stops and targets apply to RL +positions too, even though the training environment doesn't model them. + +```python +from strategies import RLStrategy +from utils.backtest import backtest_strategy +from utils.data_fetcher import fetch_market_data + +data = fetch_market_data("BTC/USD", "1h", limit=2000) +rl = RLStrategy({"parameters": {"episodes": 200, "seed": 42}}) + +bt_config = {"periods_per_year": 8760} +info = rl.before_backtest(data, bt_config) # trains on the first 70%, sets warmup to the split +result = backtest_strategy(rl, data, bt_config) +print(info["rl_training"], result["performance"]) + +rl.train(data) # or train on everything for live use +rl.save() # -> models/trained/rl_agent.json (git-ignored) +signal = rl.generate_signal({**data, "position": 0}) +``` + +All settings live under `strategies.parameters.rl_agent` in `configs/default_config.yaml`: +learning rate, gamma, episodes, the exploration schedule, seed, transaction cost, shorting, the +train/test split, persistence, and the feature definitions. Invalid values raise `ValueError`. + +On the mock market data (a random walk) the agent has nothing real to learn, so expect in-sample +gains and out-of-sample losses until real data is connected. + ## Risk and performance utilities `utils/risk.py`: @@ -232,7 +286,7 @@ These are not implemented yet: - Real market data via ccxt. - A vector-DB-backed RAG engine. - A REST / WebSocket API as specified in `docs/API.md`. -- Real multi-factor and RL strategies. +- A real multi-factor strategy. - Performance reporting. - OpenAI and local LLM providers. diff --git a/ClaudeTrader/configs/default_config.yaml b/ClaudeTrader/configs/default_config.yaml index 31e64de..ffcf943 100644 --- a/ClaudeTrader/configs/default_config.yaml +++ b/ClaudeTrader/configs/default_config.yaml @@ -52,10 +52,30 @@ strategies: multiplier: 3.0 rl_agent: - model_path: "models/trained/rl_agent.pth" - learning_rate: 0.001 - episodes: 1000 - gamma: 0.99 + # Tabular Q-learning agent (strategies/rl). All keys are optional. + model_path: "models/trained/rl_agent.json" + load_model: true # use a saved model if it matches these settings + save_model: false # write model_path after training + learning_rate: 0.1 # Q-learning step size (alpha), in (0, 1] + gamma: 0.95 # discount factor, in [0, 1) + episodes: 200 + epsilon_start: 1.0 # exploration, decayed linearly... + epsilon_end: 0.05 + epsilon_decay_fraction: 0.8 # ...over this fraction of episodes + seed: 42 # reproducible training; null for random + transaction_cost: 0.0015 # per unit of position change during training + allow_short: true + train_fraction: 0.7 # backtest() trains on this share, trades the rest + min_train_bars: 200 + features: + atr_period: 10 + supertrend_multiplier: 3.0 + rsi_period: 14 + rsi_low: 40 + rsi_high: 60 + momentum_lookback: 5 + momentum_threshold: 0.5 # x ATR% x sqrt(lookback) + volatility_lookback: 50 multi_factor: factors: diff --git a/ClaudeTrader/strategies/__init__.py b/ClaudeTrader/strategies/__init__.py index 808c3cf..a353a26 100644 --- a/ClaudeTrader/strategies/__init__.py +++ b/ClaudeTrader/strategies/__init__.py @@ -49,7 +49,11 @@ def _run_backtest( # Inherit risk parameters from the strategy where available. bt_config.setdefault('atr_period', strategy.parameters.get('atr_period', 14)) + # Strategy-specific preparation (e.g. RL training on the in-sample part). + extra = strategy.before_backtest(market_data, bt_config) or {} + result = backtest_strategy(strategy, market_data, bt_config) + result.update(extra) result.update({'strategy': strategy.name, 'symbol': symbol, 'period': period}) # Keep the equity curve out of the lightweight summary for readability. result.pop('equity_curve', None) @@ -83,7 +87,7 @@ def _period_to_bars(period: str, default: int = 500) -> int: @dataclass class StrategySignal: """Signal generated by a trading strategy""" - action: str # 'buy', 'sell', 'hold' + action: str # 'buy', 'sell', 'hold', or 'exit' (close any position) confidence: float price: float timestamp: datetime @@ -94,11 +98,27 @@ class StrategySignal: class BaseStrategy(ABC): """Base class for all trading strategies""" + # Key under ``strategies.parameters`` in the engine config. + config_key: Optional[str] = None + def __init__(self, config: Dict[str, Any]): self.config = config - self.parameters = config.get('parameters', {}) + self.parameters = self._resolve_parameters(config) self.name = self.__class__.__name__ + def _resolve_parameters(self, config: Dict[str, Any]) -> Dict[str, Any]: + """Strategy parameters, from either a direct ``parameters`` dict or the + engine config's ``strategies.parameters.`` section.""" + if config.get('parameters') is not None: + return dict(config['parameters']) + section = (config.get('strategies') or {}).get('parameters') or {} + return dict(section.get(self.config_key) or {}) + + def before_backtest(self, ohlcv: Dict[str, Any], bt_config: Dict[str, Any]) -> Optional[Dict[str, Any]]: + """Hook run before :meth:`backtest` replays data. May adjust + ``bt_config`` in place and return extra fields for the result.""" + return None + @abstractmethod def generate_signal(self, market_data: Dict[str, Any]) -> Optional[StrategySignal]: """Generate trading signal from market data""" @@ -110,8 +130,20 @@ def backtest(self, symbol: str, period: str) -> Dict[str, Any]: pass def set_parameters(self, parameters: Dict[str, Any]): - """Update strategy parameters""" + """Update strategy parameters. If the new values are rejected, the + previous parameters are restored and the error is re-raised.""" + previous = dict(self.parameters) self.parameters.update(parameters) + try: + self._on_parameters_changed() + except Exception: + self.parameters.clear() + self.parameters.update(previous) + self._on_parameters_changed() + raise + + def _on_parameters_changed(self) -> None: + """Re-read derived attributes after :meth:`set_parameters`.""" def get_parameters(self) -> Dict[str, Any]: """Get current strategy parameters""" @@ -125,8 +157,13 @@ class SuperTrendStrategy(BaseStrategy): Uses ATR-based dynamic support/resistance levels for trend following. """ + config_key = 'supertrend' + def __init__(self, config: Dict[str, Any]): super().__init__(config) + self._on_parameters_changed() + + def _on_parameters_changed(self) -> None: self.atr_period = self.parameters.get('atr_period', 10) self.multiplier = self.parameters.get('multiplier', 3.0) @@ -206,45 +243,9 @@ def _calculate_supertrend( All three returned lists are aligned to ``close`` indices (length n). ``direction`` uses 1 for uptrend and -1 for downtrend. """ - from utils.indicators import calculate_atr - - n = len(close) - atr = calculate_atr(high, low, close, self.atr_period) - - hl2 = [(high[i] + low[i]) / 2 for i in range(n)] - basic_upper = [hl2[i] + self.multiplier * atr[i] for i in range(n)] - basic_lower = [hl2[i] - self.multiplier * atr[i] for i in range(n)] - - final_upper = [0.0] * n - final_lower = [0.0] * n - direction = [1] * n - supertrend = [0.0] * n + from utils.indicators import calculate_supertrend - final_upper[0] = basic_upper[0] - final_lower[0] = basic_lower[0] - supertrend[0] = final_lower[0] - - for i in range(1, n): - # Carry the band forward unless price/band movement allows a reset. - if basic_upper[i] < final_upper[i - 1] or close[i - 1] > final_upper[i - 1]: - final_upper[i] = basic_upper[i] - else: - final_upper[i] = final_upper[i - 1] - - if basic_lower[i] > final_lower[i - 1] or close[i - 1] < final_lower[i - 1]: - final_lower[i] = basic_lower[i] - else: - final_lower[i] = final_lower[i - 1] - - prev_dir = direction[i - 1] - if prev_dir == 1: - direction[i] = -1 if close[i] < final_lower[i] else 1 - else: - direction[i] = 1 if close[i] > final_upper[i] else -1 - - supertrend[i] = final_lower[i] if direction[i] == 1 else final_upper[i] - - return supertrend, direction, atr + return calculate_supertrend(high, low, close, self.atr_period, self.multiplier) def backtest(self, symbol: str, period: str) -> Dict[str, Any]: """Backtest the SuperTrend strategy on historical/market data. @@ -260,31 +261,231 @@ class RLStrategy(BaseStrategy): """ Reinforcement Learning Strategy - Uses trained RL agent for trading decisions. + A tabular Q-learning agent (see :mod:`strategies.rl`) that maps a discrete + market state plus the current position to a target position (flat, long + or short). Signals: target long -> ``buy``, target short -> ``sell``, + target flat -> ``exit`` (or ``hold`` when already flat). + + The agent is loaded from ``model_path`` when a compatible saved model + exists, trained explicitly with :meth:`train`, or trained on the in-sample + part of the data automatically by :meth:`backtest`. Until it is trained it + only emits ``hold``. + + The current position is read from ``market_data['position']`` (1, -1 or 0; + the backtester provides it). Callers that omit it are treated as flat. """ + config_key = 'rl_agent' + def __init__(self, config: Dict[str, Any]): super().__init__(config) - self.model_path = self.parameters.get('model_path', None) - self._agent = None + self.agent = None + self.agent_source: Optional[str] = None # 'trained' or 'loaded' + self.training_summary: Optional[Dict[str, Any]] = None + self._prepared: Optional[Dict[str, Any]] = None + self._on_parameters_changed() + if self.rl_config.load_model and self.rl_config.model_path: + self.load() + + def _on_parameters_changed(self) -> None: + from strategies.rl import RLConfig + + self.rl_config = RLConfig.from_params(self.parameters) + self._prepared = None + if self.agent is not None and not self._agent_matches_config(self.agent): + logger.warning("RL parameters changed; discarding the incompatible trained agent") + self.agent = None + self.agent_source = None + + @property + def is_trained(self) -> bool: + return self.agent is not None + + # ------------------------------------------------------------------ # + # Training & persistence + # ------------------------------------------------------------------ # + def train(self, ohlcv: Dict[str, Any], end: Optional[int] = None) -> Dict[str, Any]: + """Train the agent on bars ``[0, end)`` of ``ohlcv``. + + Returns a training summary. Saves to ``model_path`` if ``save_model``. + """ + from strategies.rl import compute_market_states, train_agent + + states = compute_market_states(ohlcv, self.rl_config.features) + result = train_agent(ohlcv, self.rl_config, end=end, market_states=states) + self.agent = result.agent + self.agent_source = 'trained' + self.training_summary = result.summary() + self._cache_states(ohlcv, states) + if self.rl_config.save_model and self.rl_config.model_path: + self.save() + return self.training_summary + + def save(self, path: Optional[str] = None) -> str: + if self.agent is None: + raise RuntimeError("RL agent is not trained; nothing to save") + target = path or self.rl_config.model_path + if not target: + raise ValueError("no model_path configured") + return str(self.agent.save(target)) + + def load(self, path: Optional[str] = None) -> bool: + """Load a saved agent. Returns False (keeping the current agent) if the + file is missing, unreadable, or was trained with different settings.""" + from pathlib import Path + from strategies.rl import QLearningAgent + + source = path or self.rl_config.model_path + if not source or not Path(source).exists(): + return False + try: + agent = QLearningAgent.load(source) + except (OSError, ValueError, KeyError) as e: + logger.warning(f"Could not load RL model from {source}: {e}") + return False + if not self._agent_matches_config(agent): + logger.warning(f"RL model at {source} was trained with different settings; ignoring it") + return False + self.agent = agent + self.agent_source = 'loaded' + logger.info(f"Loaded RL agent from {source}") + return True + + def _agent_matches_config(self, agent) -> bool: + return ( + agent.metadata.get('features') == self.rl_config.features.to_dict() + and agent.allow_short == self.rl_config.allow_short + ) + + # ------------------------------------------------------------------ # + # Signals + # ------------------------------------------------------------------ # + def prepare(self, ohlcv: Dict[str, Any]) -> None: + """Precompute market states for a full series. Because states are + causal, later windows that are prefixes of this series reuse them.""" + from strategies.rl import compute_market_states + + cached = self._prepared + if cached is not None and list(ohlcv.get('close', [])) == cached['close'] \ + and list(ohlcv.get('high', [])) == cached['high'] \ + and list(ohlcv.get('low', [])) == cached['low']: + return + self._cache_states(ohlcv, compute_market_states(ohlcv, self.rl_config.features)) + + def _cache_states(self, ohlcv: Dict[str, Any], states: List[Optional[int]]) -> None: + self._prepared = { + 'close': list(ohlcv.get('close', [])), + 'high': list(ohlcv.get('high', [])), + 'low': list(ohlcv.get('low', [])), + 'states': states, + } + + def _market_state(self, market_data: Dict[str, Any]) -> Optional[int]: + close = market_data.get('close', []) + n = len(close) + cached = self._prepared + if ( + cached is not None + and 0 < n <= len(cached['close']) + and list(close) == cached['close'][:n] + and list(market_data.get('high', close)) == cached['high'][:n] + and list(market_data.get('low', close)) == cached['low'][:n] + ): + return cached['states'][n - 1] + + from strategies.rl import compute_market_states + + return compute_market_states(market_data, self.rl_config.features)[-1] if n else None def generate_signal(self, market_data: Dict[str, Any]) -> Optional[StrategySignal]: - """Generate signal using RL agent""" + """Generate a signal from the trained agent's greedy policy.""" + from strategies.rl import ACTION_NAMES, FLAT, LONG, SHORT, agent_state, decode_market_state + + close = market_data.get('close') or [0] + price = close[-1] + position = int(market_data.get('position', 0) or 0) + position = max(-1, min(1, position)) + if position == -1 and not self.rl_config.allow_short: + position = 0 + + def hold(reason: str, indicators: Optional[Dict[str, Any]] = None) -> StrategySignal: + return StrategySignal( + action='hold', confidence=0.0, price=price, timestamp=datetime.now(), + indicators=indicators or {'position': position}, reasoning=reason, + ) + + if self.agent is None: + return hold("RL agent is not trained yet (call train() or backtest())") + + market_state = self._market_state(market_data) + if market_state is None: + return hold("Not enough history for RL state features") - # Placeholder for RL agent - # Future enhancement: Load and use actual trained RL model + state = agent_state(market_state, position) + action = self.agent.greedy_action(state) + confidence = self.agent.confidence(state) + q = self.agent.q_values(state) + + if action == LONG: + signal_action = 'buy' + elif action == SHORT: + signal_action = 'sell' + else: + signal_action = 'exit' if position != 0 else 'hold' + + indicators: Dict[str, Any] = { + 'position': position, + 'target': ACTION_NAMES[action], + 'market_state': market_state, + 'state_visits': int(self.agent.visits[state].sum()), + **{f'q_{ACTION_NAMES[a]}': v for a, v in q.items()}, + **decode_market_state(market_state), + } + if indicators['state_visits'] == 0: + return hold("State never seen in training; staying put", indicators) return StrategySignal( - action='hold', - confidence=0.6, - price=market_data.get('close', [0])[-1], + action=signal_action, + confidence=confidence, + price=price, timestamp=datetime.now(), - indicators={'rl_score': 0.6}, - reasoning="RL agent decision (placeholder)" + indicators=indicators, + reasoning=f"RL agent targets {ACTION_NAMES[action]} (Q-margin confidence {confidence:.2f})", ) + # ------------------------------------------------------------------ # + # Backtesting + # ------------------------------------------------------------------ # + def before_backtest(self, ohlcv: Dict[str, Any], bt_config: Dict[str, Any]) -> Optional[Dict[str, Any]]: + """Train on the first ``train_fraction`` of the data and only trade the + remaining, unseen part. + + An agent loaded from ``model_path`` is used as-is (trading still starts + at the split). An agent trained earlier in memory is always retrained + here, since it may have seen this backtest's out-of-sample bars or a + different market entirely. + """ + n = len(ohlcv.get('close', [])) + split = int(n * self.rl_config.train_fraction) + info: Dict[str, Any] = {'rl_split_index': split} + + if self.agent_source != 'loaded': + if split < self.rl_config.min_train_bars: + raise ValueError( + f"Not enough data to train the RL agent: {split} in-sample bars, " + f"need {self.rl_config.min_train_bars}" + ) + info['rl_training'] = self.train(ohlcv, end=split) + info['rl_model_source'] = 'trained_in_sample' + else: + info['rl_model_source'] = 'loaded' + + bt_config['warmup'] = max(int(bt_config.get('warmup') or 0), split) + bt_config.setdefault('allow_short', self.rl_config.allow_short) + return info + def backtest(self, symbol: str, period: str) -> Dict[str, Any]: - """Backtest RL strategy using the event-driven engine.""" + """Train in-sample and backtest out-of-sample with the event-driven engine.""" return _run_backtest(self, symbol, period, self.config) @@ -295,6 +496,8 @@ class MultiFactorStrategy(BaseStrategy): Combines multiple technical indicators and factors. """ + config_key = 'multi_factor' + def __init__(self, config: Dict[str, Any]): super().__init__(config) self.factors = self.parameters.get('factors', ['trend', 'momentum', 'volatility']) @@ -364,6 +567,8 @@ class HybridStrategy(BaseStrategy): Combines multiple strategies with ensemble logic. """ + config_key = 'hybrid' + def __init__(self, config: Dict[str, Any]): super().__init__(config) self.strategies = [ diff --git a/ClaudeTrader/strategies/rl/__init__.py b/ClaudeTrader/strategies/rl/__init__.py new file mode 100644 index 0000000..3164dcb --- /dev/null +++ b/ClaudeTrader/strategies/rl/__init__.py @@ -0,0 +1,31 @@ +""" +Reinforcement-learning components for ClaudeTrader. + +* :mod:`.features` - causal, discrete market-state features +* :mod:`.environment` - trading environment with backtester-consistent semantics +* :mod:`.agent` - tabular Q-learning agent with JSON persistence +* :mod:`.trainer` - :class:`RLConfig` and :func:`train_agent` + +The strategy wrapper lives in :class:`strategies.RLStrategy`. +""" + +from .agent import QLearningAgent +from .environment import ACTION_NAMES, FLAT, LONG, SHORT, TradingEnvironment, agent_state +from .features import FeatureConfig, compute_market_states, decode_market_state +from .trainer import RLConfig, TrainingResult, train_agent + +__all__ = [ + "ACTION_NAMES", + "FLAT", + "LONG", + "SHORT", + "FeatureConfig", + "QLearningAgent", + "RLConfig", + "TradingEnvironment", + "TrainingResult", + "agent_state", + "compute_market_states", + "decode_market_state", + "train_agent", +] diff --git a/ClaudeTrader/strategies/rl/agent.py b/ClaudeTrader/strategies/rl/agent.py new file mode 100644 index 0000000..0c067e2 --- /dev/null +++ b/ClaudeTrader/strategies/rl/agent.py @@ -0,0 +1,150 @@ +""" +Tabular Q-learning agent. + +The state space is small (market state x current position), so a Q-table is +learned exactly, needs no deep-learning dependency, trains in well under a +second on thousands of bars, and is fully reproducible for a given seed. The +table is saved as JSON alongside the feature configuration it was trained with. +""" + +from __future__ import annotations + +import json +import logging +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, Optional + +import numpy as np + +from .environment import N_ACTIONS, N_POSITIONS, valid_actions +from .features import N_MARKET_STATES + +logger = logging.getLogger(__name__) + +MODEL_FORMAT_VERSION = 1 + + +class QLearningAgent: + """Epsilon-greedy tabular Q-learning agent.""" + + def __init__( + self, + learning_rate: float = 0.1, + gamma: float = 0.95, + allow_short: bool = True, + n_states: int = N_MARKET_STATES * N_POSITIONS, + seed: Optional[int] = None, + ): + if not 0 < learning_rate <= 1: + raise ValueError("learning_rate must be in (0, 1]") + if not 0 <= gamma < 1: + raise ValueError("gamma must be in [0, 1)") + self.learning_rate = learning_rate + self.gamma = gamma + self.allow_short = allow_short + self.n_states = n_states + self.actions = valid_actions(allow_short) + self.q = np.zeros((n_states, N_ACTIONS)) + self.visits = np.zeros((n_states, N_ACTIONS), dtype=np.int64) + self.rng = np.random.default_rng(seed) + self.metadata: Dict[str, Any] = {} + + # ------------------------------------------------------------------ # + # Acting + # ------------------------------------------------------------------ # + def greedy_action(self, state: int) -> int: + """Best known action; ties (including unseen states) prefer the first + valid action, i.e. staying flat.""" + values = self.q[state, self.actions] + return self.actions[int(np.argmax(values))] + + def act(self, state: int, epsilon: float) -> int: + if self.rng.random() < epsilon: + return int(self.rng.choice(self.actions)) + return self.greedy_action(state) + + def confidence(self, state: int) -> float: + """Confidence in the greedy action, in [0, 1]. + + Zero for a state never visited in training; otherwise the margin of the + best action over the runner-up, relative to their magnitudes. + """ + if self.visits[state].sum() == 0: + return 0.0 + values = np.sort(self.q[state, self.actions])[::-1] + best, second = values[0], values[1] + scale = abs(best) + abs(second) + if scale <= 1e-12: + return 0.0 + return float(min(max((best - second) / scale, 0.0), 1.0)) + + def q_values(self, state: int) -> Dict[int, float]: + return {a: float(self.q[state, a]) for a in self.actions} + + # ------------------------------------------------------------------ # + # Learning + # ------------------------------------------------------------------ # + def update(self, state: int, action: int, reward: float, next_state: Optional[int]) -> None: + """One Q-learning update. ``next_state`` is ``None`` at episode end.""" + target = reward + if next_state is not None: + target += self.gamma * float(np.max(self.q[next_state, self.actions])) + self.q[state, action] += self.learning_rate * (target - self.q[state, action]) + self.visits[state, action] += 1 + + # ------------------------------------------------------------------ # + # Persistence + # ------------------------------------------------------------------ # + def to_dict(self) -> Dict[str, Any]: + return { + "format_version": MODEL_FORMAT_VERSION, + "algorithm": "tabular_q_learning", + "learning_rate": self.learning_rate, + "gamma": self.gamma, + "allow_short": self.allow_short, + "n_states": self.n_states, + "q": self.q.tolist(), + "visits": self.visits.tolist(), + "metadata": self.metadata, + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "QLearningAgent": + if data.get("format_version") != MODEL_FORMAT_VERSION: + raise ValueError(f"unsupported model format: {data.get('format_version')}") + agent = cls( + learning_rate=data["learning_rate"], + gamma=data["gamma"], + allow_short=data["allow_short"], + n_states=data["n_states"], + ) + q = np.asarray(data["q"], dtype=float) + visits = np.asarray(data["visits"], dtype=np.int64) + if q.shape != agent.q.shape or visits.shape != agent.visits.shape: + raise ValueError("model table shape does not match the state/action space") + agent.q, agent.visits = q, visits + agent.metadata = dict(data.get("metadata", {})) + return agent + + def save(self, path: str) -> Path: + out = Path(path) + out.parent.mkdir(parents=True, exist_ok=True) + self.metadata.setdefault("saved_at", datetime.now(timezone.utc).isoformat()) + tmp = out.with_suffix(out.suffix + ".tmp") + tmp.write_text(json.dumps(self.to_dict())) + tmp.replace(out) # atomic: never leaves a half-written model + logger.info(f"Saved RL agent to {out}") + return out + + @classmethod + def load(cls, path: str) -> "QLearningAgent": + return cls.from_dict(json.loads(Path(path).read_text())) + + +def epsilon_schedule(episode: int, episodes: int, start: float, end: float, decay_fraction: float) -> float: + """Linear decay from ``start`` to ``end`` over the first ``decay_fraction`` + of training, then constant.""" + decay_episodes = max(1, int(episodes * decay_fraction)) + progress = min(episode / decay_episodes, 1.0) + return start + (end - start) * progress diff --git a/ClaudeTrader/strategies/rl/environment.py b/ClaudeTrader/strategies/rl/environment.py new file mode 100644 index 0000000..dd4873d --- /dev/null +++ b/ClaudeTrader/strategies/rl/environment.py @@ -0,0 +1,120 @@ +""" +Trading environment for the RL agent. + +A minimal, gym-style episodic environment over a precomputed sequence of market +states. Actions are *target positions*: + +* ``FLAT`` (0) - hold no position +* ``LONG`` (1) - hold a long position +* ``SHORT`` (2) - hold a short position (masked out when shorting is disabled) + +At bar ``t`` the agent picks a target position, which is then held from the +close of bar ``t`` to the close of bar ``t + 1``. The reward is the log return +earned by that position minus a proportional cost for every unit of position +change. This mirrors how the backtester fills signals at the bar close. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass +from typing import List, Optional, Sequence, Tuple + +FLAT, LONG, SHORT = 0, 1, 2 +N_ACTIONS = 3 +ACTION_NAMES = {FLAT: "flat", LONG: "long", SHORT: "short"} +ACTION_TO_POSITION = {FLAT: 0, LONG: 1, SHORT: -1} +POSITION_TO_INDEX = {0: FLAT, 1: LONG, -1: SHORT} +N_POSITIONS = 3 + + +def agent_state(market_state: int, position: int) -> int: + """Combine a market state with the current position (-1, 0, 1).""" + return market_state * N_POSITIONS + POSITION_TO_INDEX[position] + + +def valid_actions(allow_short: bool) -> List[int]: + return [FLAT, LONG, SHORT] if allow_short else [FLAT, LONG] + + +@dataclass +class StepResult: + state: Optional[int] + reward: float + done: bool + + +class TradingEnvironment: + """Episodic environment over bars ``[start, end)`` of a price series.""" + + def __init__( + self, + close: Sequence[float], + market_states: Sequence[Optional[int]], + transaction_cost: float = 0.0015, + allow_short: bool = True, + start: Optional[int] = None, + end: Optional[int] = None, + ): + if len(close) != len(market_states): + raise ValueError("close and market_states must have the same length") + self.close = [float(c) for c in close] + self.market_states = list(market_states) + self.transaction_cost = transaction_cost + self.allow_short = allow_short + + first_valid = next((i for i, s in enumerate(self.market_states) if s is not None), None) + if first_valid is None: + raise ValueError("no bar has a defined market state (series too short for warmup)") + self.start = max(first_valid, start or 0) + self.end = min(len(self.close), end if end is not None else len(self.close)) + if self.end - self.start < 2: + raise ValueError("need at least two bars with defined states to form an episode") + + self.t = self.start + self.position = 0 + + @property + def num_steps(self) -> int: + return self.end - 1 - self.start + + def reset(self) -> int: + self.t = self.start + self.position = 0 + return self.current_state() + + def current_state(self) -> int: + return agent_state(self.market_states[self.t], self.position) + + def step(self, action: int) -> StepResult: + if action == SHORT and not self.allow_short: + raise ValueError("short action is disabled for this environment") + + target = ACTION_TO_POSITION[action] + cost = self.transaction_cost * abs(target - self.position) + prev, nxt = self.close[self.t], self.close[self.t + 1] + log_return = math.log(nxt / prev) if prev > 0 and nxt > 0 else 0.0 + reward = target * log_return - cost + + self.position = target + self.t += 1 + done = self.t >= self.end - 1 + state = None if done else self.current_state() + return StepResult(state=state, reward=reward, done=done) + + def run_policy(self, policy) -> Tuple[float, int]: + """Run a deterministic ``policy(state) -> action`` for one episode. + + Returns: + (total reward, number of position changes) + """ + state = self.reset() + total, changes = 0.0, 0 + while True: + before = self.position + result = self.step(policy(state)) + total += result.reward + changes += int(self.position != before) + if result.done: + return total, changes + state = result.state diff --git a/ClaudeTrader/strategies/rl/features.py b/ClaudeTrader/strategies/rl/features.py new file mode 100644 index 0000000..1bd8e25 --- /dev/null +++ b/ClaudeTrader/strategies/rl/features.py @@ -0,0 +1,138 @@ +""" +State features for the RL trading agent. + +Turns an OHLCV series into one discrete market state per bar. Every state is +causal: the state at bar ``t`` depends only on bars ``0..t``, so the same +series prefix always produces the same states (no look-ahead). Bars that do not +yet have enough history map to ``None``. + +The market state combines four bucketed features: + +* ``trend`` - SuperTrend direction (down / up) +* ``rsi`` - RSI zone (weak / neutral / strong) +* ``momentum`` - N-bar return relative to current volatility (down / flat / up) +* ``volatility`` - ATR% versus its rolling median (calm / elevated) +""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass, fields +from typing import Any, Dict, List, Optional, Sequence + +import numpy as np + +from utils.indicators import calculate_atr, calculate_rsi, calculate_supertrend + +TREND_LEVELS = 2 +RSI_LEVELS = 3 +MOMENTUM_LEVELS = 3 +VOLATILITY_LEVELS = 2 +N_MARKET_STATES = TREND_LEVELS * RSI_LEVELS * MOMENTUM_LEVELS * VOLATILITY_LEVELS + + +@dataclass +class FeatureConfig: + """Parameters that define the state space. A trained agent is only valid + for the feature configuration it was trained with.""" + + atr_period: int = 10 + supertrend_multiplier: float = 3.0 + rsi_period: int = 14 + rsi_low: float = 40.0 + rsi_high: float = 60.0 + momentum_lookback: int = 5 + momentum_threshold: float = 0.5 # in units of ATR% * sqrt(lookback) + volatility_lookback: int = 50 + + @classmethod + def from_params(cls, params: Dict[str, Any]) -> "FeatureConfig": + names = {f.name for f in fields(cls)} + return cls(**{k: v for k, v in params.items() if k in names}) + + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + + @property + def warmup(self) -> int: + """First bar index at which every feature is defined.""" + return max( + self.atr_period + 1, + self.rsi_period + 1, + self.momentum_lookback, + self.volatility_lookback, + ) + + +def encode_market_state(trend: int, rsi_zone: int, momentum: int, volatility: int) -> int: + """Combine bucketed features into a single index in ``[0, N_MARKET_STATES)``.""" + return ((trend * RSI_LEVELS + rsi_zone) * MOMENTUM_LEVELS + momentum) * VOLATILITY_LEVELS + volatility + + +def decode_market_state(state: int) -> Dict[str, int]: + """Inverse of :func:`encode_market_state`.""" + state, volatility = divmod(state, VOLATILITY_LEVELS) + state, momentum = divmod(state, MOMENTUM_LEVELS) + trend, rsi_zone = divmod(state, RSI_LEVELS) + return {"trend": trend, "rsi": rsi_zone, "momentum": momentum, "volatility": volatility} + + +def compute_market_states( + ohlcv: Dict[str, Sequence[float]], + config: Optional[FeatureConfig] = None, +) -> List[Optional[int]]: + """Compute the discrete market state for every bar of ``ohlcv``. + + Args: + ohlcv: Dict with ``high``, ``low`` and ``close`` lists of equal length. + config: Feature parameters (defaults if omitted). + + Returns: + A list aligned with ``close``: an int state, or ``None`` during warmup. + """ + config = config or FeatureConfig() + close = [float(x) for x in ohlcv.get("close", [])] + high = [float(x) for x in ohlcv.get("high", close)] + low = [float(x) for x in ohlcv.get("low", close)] + n = len(close) + if n == 0: + return [] + + _, direction, _ = calculate_supertrend( + high, low, close, config.atr_period, config.supertrend_multiplier + ) + rsi = calculate_rsi(close, config.rsi_period) + atr = calculate_atr(high, low, close, config.atr_period) + + close_arr = np.asarray(close) + atr_pct = np.divide( + np.asarray(atr), close_arr, out=np.zeros(n), where=close_arr > 0 + ) + + states: List[Optional[int]] = [None] * n + k = config.momentum_lookback + for t in range(config.warmup, n): + trend = 1 if direction[t] == 1 else 0 + + if rsi[t] < config.rsi_low: + rsi_zone = 0 + elif rsi[t] > config.rsi_high: + rsi_zone = 2 + else: + rsi_zone = 1 + + threshold = config.momentum_threshold * atr_pct[t] * np.sqrt(k) + past = close[t - k] + ret = close[t] / past - 1 if past > 0 else 0.0 + if ret > threshold: + momentum = 2 + elif ret < -threshold: + momentum = 0 + else: + momentum = 1 + + window = atr_pct[t - config.volatility_lookback + 1 : t + 1] + volatility = 1 if atr_pct[t] > np.median(window) else 0 + + states[t] = encode_market_state(trend, rsi_zone, momentum, volatility) + + return states diff --git a/ClaudeTrader/strategies/rl/trainer.py b/ClaudeTrader/strategies/rl/trainer.py new file mode 100644 index 0000000..ccd2f7a --- /dev/null +++ b/ClaudeTrader/strategies/rl/trainer.py @@ -0,0 +1,193 @@ +""" +Training entry point for the RL agent. + +:class:`RLConfig` collects every tunable variable (learning, exploration, +costs, feature definitions, persistence) with validated defaults, and +:func:`train_agent` trains a :class:`QLearningAgent` on an OHLCV series. +""" + +from __future__ import annotations + +import logging +from dataclasses import asdict, dataclass, field, fields +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional, Sequence + +from .agent import QLearningAgent, epsilon_schedule +from .environment import TradingEnvironment +from .features import FeatureConfig, compute_market_states + +logger = logging.getLogger(__name__) + + +@dataclass +class RLConfig: + """All RL strategy variables, with defaults.""" + + # Learning + learning_rate: float = 0.1 + gamma: float = 0.95 + episodes: int = 200 + epsilon_start: float = 1.0 + epsilon_end: float = 0.05 + epsilon_decay_fraction: float = 0.8 + seed: Optional[int] = 42 + + # Trading environment + transaction_cost: float = 0.0015 # per unit of position change (commission + slippage) + allow_short: bool = True + + # Train/test split used by backtest(): train on the first fraction only + train_fraction: float = 0.7 + min_train_bars: int = 200 + + # Persistence + model_path: Optional[str] = "models/trained/rl_agent.json" + load_model: bool = True # load model_path at startup if it exists and matches + save_model: bool = False # write model_path after training + + features: FeatureConfig = field(default_factory=FeatureConfig) + + @classmethod + def from_params(cls, params: Optional[Dict[str, Any]]) -> "RLConfig": + """Build from a flat parameter dict (as in ``default_config.yaml``). + + Feature parameters may be given flat or under a ``features`` key. + Unknown keys are ignored with a warning. + """ + params = dict(params or {}) + own = {f.name for f in fields(cls)} - {"features"} + feature_names = {f.name for f in fields(FeatureConfig)} + + feature_params = dict(params.pop("features", None) or {}) + kwargs: Dict[str, Any] = {} + for key, value in params.items(): + if key in own: + kwargs[key] = value + elif key in feature_names: + feature_params[key] = value + else: + logger.warning(f"Ignoring unknown rl_agent parameter: {key}") + + config = cls(features=FeatureConfig.from_params(feature_params), **kwargs) + config.validate() + return config + + def validate(self) -> None: + checks = [ + (0 < self.learning_rate <= 1, "learning_rate must be in (0, 1]"), + (0 <= self.gamma < 1, "gamma must be in [0, 1)"), + (self.episodes >= 1, "episodes must be >= 1"), + (0 <= self.epsilon_end <= self.epsilon_start <= 1, "need 0 <= epsilon_end <= epsilon_start <= 1"), + (0 < self.epsilon_decay_fraction <= 1, "epsilon_decay_fraction must be in (0, 1]"), + (self.transaction_cost >= 0, "transaction_cost must be >= 0"), + (0 < self.train_fraction < 1, "train_fraction must be in (0, 1)"), + (self.min_train_bars >= 2, "min_train_bars must be >= 2"), + (self.features.momentum_lookback >= 1, "momentum_lookback must be >= 1"), + (self.features.volatility_lookback >= 2, "volatility_lookback must be >= 2"), + (self.features.rsi_low < self.features.rsi_high, "rsi_low must be below rsi_high"), + ] + for ok, message in checks: + if not ok: + raise ValueError(f"Invalid RL config: {message}") + + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + + +@dataclass +class TrainingResult: + agent: QLearningAgent + episode_rewards: List[float] + train_start: int + train_end: int + final_greedy_reward: float + final_greedy_trades: int + + def summary(self) -> Dict[str, Any]: + rewards = self.episode_rewards + tail = rewards[-max(1, len(rewards) // 10):] + return { + "episodes": len(rewards), + "train_bars": self.train_end - self.train_start, + "train_range": [self.train_start, self.train_end], + "mean_reward_last_10pct": sum(tail) / len(tail), + "greedy_log_return": self.final_greedy_reward, + "greedy_position_changes": self.final_greedy_trades, + } + + +def train_agent( + ohlcv: Dict[str, Sequence[float]], + config: Optional[RLConfig] = None, + end: Optional[int] = None, + market_states: Optional[Sequence[Optional[int]]] = None, +) -> TrainingResult: + """Train a Q-learning agent on bars ``[0, end)`` of ``ohlcv``. + + Args: + ohlcv: OHLCV dict (``high``/``low``/``close`` required). + config: RL configuration (defaults if omitted). + end: Exclusive bar index to stop training at (defaults to all bars). + Pass the train/test split here so evaluation data is never seen. + market_states: Precomputed states for ``ohlcv`` (computed if omitted). + + Returns: + A :class:`TrainingResult` with the trained agent and statistics. + """ + config = config or RLConfig() + config.validate() + close = list(ohlcv.get("close", [])) + if market_states is None: + market_states = compute_market_states(ohlcv, config.features) + + env = TradingEnvironment( + close, + market_states, + transaction_cost=config.transaction_cost, + allow_short=config.allow_short, + end=end, + ) + agent = QLearningAgent( + learning_rate=config.learning_rate, + gamma=config.gamma, + allow_short=config.allow_short, + seed=config.seed, + ) + + episode_rewards: List[float] = [] + for episode in range(config.episodes): + epsilon = epsilon_schedule( + episode, config.episodes, config.epsilon_start, config.epsilon_end, config.epsilon_decay_fraction + ) + state = env.reset() + total = 0.0 + while True: + action = agent.act(state, epsilon) + result = env.step(action) + agent.update(state, action, result.reward, result.state) + total += result.reward + if result.done: + break + state = result.state + episode_rewards.append(total) + + greedy_reward, greedy_trades = env.run_policy(agent.greedy_action) + agent.metadata.update({ + "trained_at": datetime.now(timezone.utc).isoformat(), + "features": config.features.to_dict(), + "transaction_cost": config.transaction_cost, + "episodes": config.episodes, + "train_bars": env.end - env.start, + }) + + result = TrainingResult( + agent=agent, + episode_rewards=episode_rewards, + train_start=env.start, + train_end=env.end, + final_greedy_reward=greedy_reward, + final_greedy_trades=greedy_trades, + ) + logger.info(f"RL training complete: {result.summary()}") + return result diff --git a/ClaudeTrader/tests/test_rl.py b/ClaudeTrader/tests/test_rl.py new file mode 100644 index 0000000..da5a729 --- /dev/null +++ b/ClaudeTrader/tests/test_rl.py @@ -0,0 +1,287 @@ +"""Tests for the RL strategy stack (features, environment, agent, strategy).""" + +import math + +import numpy as np +import pytest + +from strategies import RLStrategy, SuperTrendStrategy, get_strategy +from strategies.rl import ( + FLAT, + LONG, + SHORT, + FeatureConfig, + QLearningAgent, + RLConfig, + TradingEnvironment, + agent_state, + compute_market_states, + decode_market_state, + train_agent, +) +from strategies.rl.features import N_MARKET_STATES, encode_market_state +from utils.backtest import Backtester, backtest_strategy + +FAST = {"episodes": 30, "load_model": False} + + +def _ohlcv(n=400, seed=3, drift=0.0005, vol=0.02): + rng = np.random.default_rng(seed) + close = list(100 * np.cumprod(1 + rng.normal(drift, vol, n))) + high = [c * (1 + abs(rng.normal(0, 0.004))) for c in close] + low = [c * (1 - abs(rng.normal(0, 0.004))) for c in close] + return {"open": list(close), "high": high, "low": low, "close": close} + + +def _trend(n=300, step=0.01): + close = [100 * (1 + step) ** i for i in range(n)] + return {"high": [c * 1.002 for c in close], "low": [c * 0.998 for c in close], "close": close} + + +# ---------------------------------------------------------------- features +def test_market_states_are_causal(): + data = _ohlcv(300) + full = compute_market_states(data) + for cut in (60, 120, 299): + prefix = {k: v[:cut] for k, v in data.items()} + assert compute_market_states(prefix) == full[:cut] + + +def test_market_states_warmup_and_range(): + config = FeatureConfig() + states = compute_market_states(_ohlcv(200), config) + assert all(s is None for s in states[: config.warmup]) + assert all(0 <= s < N_MARKET_STATES for s in states[config.warmup:]) + + +def test_state_encoding_round_trip(): + for state in range(N_MARKET_STATES): + parts = decode_market_state(state) + assert encode_market_state(parts["trend"], parts["rsi"], parts["momentum"], parts["volatility"]) == state + + +def test_uptrend_reads_as_uptrend(): + states = [s for s in compute_market_states(_trend()) if s is not None] + assert all(decode_market_state(s)["trend"] == 1 for s in states[-50:]) + + +# ------------------------------------------------------------- environment +def test_environment_reward_and_costs(): + close = [100.0, 110.0, 99.0] + env = TradingEnvironment(close, [0, 0, 0], transaction_cost=0.01) + env.reset() + step = env.step(LONG) + assert step.reward == pytest.approx(math.log(1.1) - 0.01) + assert env.position == 1 and not step.done + step = env.step(FLAT) # closing costs, earns nothing + assert step.reward == pytest.approx(-0.01) + assert step.done and step.state is None + env.reset() + env.step(SHORT) + assert env.position == -1 + + +def test_environment_rejects_short_when_disabled(): + env = TradingEnvironment([1.0, 2.0, 3.0], [0, 0, 0], allow_short=False) + env.reset() + with pytest.raises(ValueError): + env.step(SHORT) + + +def test_environment_requires_defined_states(): + with pytest.raises(ValueError): + TradingEnvironment([1.0, 2.0], [None, None]) + + +# ------------------------------------------------------------------- agent +def test_agent_learns_to_be_long_in_a_steady_uptrend(): + result = train_agent(_trend(), RLConfig(episodes=50, transaction_cost=0.001)) + assert result.final_greedy_reward > 0 + env_states = [s for s in compute_market_states(_trend()) if s is not None] + assert result.agent.greedy_action(agent_state(env_states[-1], 1)) == LONG + + +def test_training_is_reproducible_for_a_seed(): + data = _ohlcv(300) + a = train_agent(data, RLConfig(episodes=10, seed=7)).agent + b = train_agent(data, RLConfig(episodes=10, seed=7)).agent + assert np.array_equal(a.q, b.q) + + +def test_training_end_limits_data_seen(): + data = _ohlcv(400) + result = train_agent(data, RLConfig(episodes=5), end=250) + assert result.train_end == 250 + shorter = train_agent({k: v[:250] for k, v in data.items()}, RLConfig(episodes=5)).agent + assert np.array_equal(result.agent.q, shorter.q) + + +def test_agent_without_shorts_never_shorts(): + agent = train_agent(_ohlcv(300), RLConfig(episodes=10, allow_short=False)).agent + assert all(agent.greedy_action(s) != SHORT for s in range(agent.n_states)) + assert not agent.visits[:, SHORT].any() + + +def test_agent_save_load_round_trip(tmp_path): + agent = train_agent(_ohlcv(300), RLConfig(episodes=5)).agent + path = agent.save(str(tmp_path / "nested" / "agent.json")) + loaded = QLearningAgent.load(str(path)) + assert np.array_equal(loaded.q, agent.q) + assert np.array_equal(loaded.visits, agent.visits) + assert loaded.metadata["features"] == FeatureConfig().to_dict() + + +def test_agent_rejects_bad_hyperparameters(): + with pytest.raises(ValueError): + QLearningAgent(learning_rate=0) + with pytest.raises(ValueError): + QLearningAgent(gamma=1.0) + + +# ------------------------------------------------------------------ config +def test_rl_config_from_params_flat_and_nested_features(): + cfg = RLConfig.from_params({"episodes": 12, "rsi_period": 21, "features": {"momentum_lookback": 3}}) + assert cfg.episodes == 12 + assert cfg.features.rsi_period == 21 + assert cfg.features.momentum_lookback == 3 + + +@pytest.mark.parametrize("bad", [ + {"learning_rate": 1.5}, + {"gamma": 1.0}, + {"train_fraction": 1.0}, + {"epsilon_start": 0.1, "epsilon_end": 0.5}, + {"rsi_low": 70, "rsi_high": 30}, +]) +def test_rl_config_validation(bad): + with pytest.raises(ValueError): + RLConfig.from_params(bad) + + +def test_engine_config_parameters_reach_strategies(): + config = {"strategies": {"parameters": { + "rl_agent": {"episodes": 17, "load_model": False}, + "supertrend": {"atr_period": 7, "multiplier": 2.5}, + }}} + assert get_strategy("rl_agent", config).rl_config.episodes == 17 + st = get_strategy("supertrend", config) + assert (st.atr_period, st.multiplier) == (7, 2.5) + + +def test_set_parameters_updates_derived_attributes(): + st = SuperTrendStrategy({"parameters": {}}) + st.set_parameters({"atr_period": 21}) + assert st.atr_period == 21 + rl = RLStrategy({"parameters": dict(FAST)}) + rl.set_parameters({"episodes": 3}) + assert rl.rl_config.episodes == 3 + + +# ---------------------------------------------------------------- strategy +def test_untrained_strategy_holds(): + sig = RLStrategy({"parameters": dict(FAST)}).generate_signal(_ohlcv(100)) + assert sig.action == "hold" and sig.confidence == 0.0 + + +def test_strategy_signal_mapping_uses_position(): + rl = RLStrategy({"parameters": dict(FAST)}) + data = _ohlcv(300) + rl.train(data) + market_state = compute_market_states(data)[-1] + + rl.agent.q[:] = 0.0 + rl.agent.visits[:] = 1 + rl.agent.q[agent_state(market_state, 1), FLAT] = 1.0 # long -> wants flat + rl.agent.q[agent_state(market_state, 0), SHORT] = 1.0 # flat -> wants short + + assert rl.generate_signal({**data, "position": 1}).action == "exit" + assert rl.generate_signal({**data, "position": 0}).action == "sell" + + +def test_strategy_save_and_load_compatibility(tmp_path): + path = str(tmp_path / "rl.json") + params = {**FAST, "model_path": path, "save_model": True} + trained = RLStrategy({"parameters": params}) + trained.train(_ohlcv(300)) + + reloaded = RLStrategy({"parameters": {**params, "load_model": True}}) + assert reloaded.is_trained + assert np.array_equal(reloaded.agent.q, trained.agent.q) + + different = RLStrategy({"parameters": {**params, "load_model": True, "rsi_period": 21}}) + assert not different.is_trained # trained with other features: ignored + + +def test_strategy_backtest_trades_only_out_of_sample(): + data = _ohlcv(600, seed=11) + rl = RLStrategy({"parameters": dict(FAST)}) + bt_config = {} + info = rl.before_backtest(data, bt_config) + assert info["rl_model_source"] == "trained_in_sample" + split = info["rl_split_index"] + assert split == int(600 * 0.7) and bt_config["warmup"] == split + + out = backtest_strategy(rl, data, bt_config) + assert all(t["entry_index"] >= split for t in out["trades"]) + # Deterministic for a fixed seed. + again = RLStrategy({"parameters": dict(FAST)}) + again.before_backtest(data, {}) + assert backtest_strategy(again, data, bt_config)["final_capital"] == out["final_capital"] + + +def test_strategy_requires_enough_training_data(): + with pytest.raises(ValueError): + RLStrategy({"parameters": dict(FAST)}).before_backtest(_ohlcv(100), {}) + + +def test_cached_states_match_fresh_computation(): + data = _ohlcv(300) + rl = RLStrategy({"parameters": dict(FAST)}) + rl.prepare(data) + fresh = compute_market_states(data) + for n in (60, 200, 300): + assert rl._market_state({k: v[:n] for k, v in data.items()}) == fresh[n - 1] + + +# ------------------------------------------------- backtester contract +def test_backtester_exit_action_and_position_in_window(): + seen = [] + + def signal(window): + i = len(window["close"]) - 1 + seen.append((i, window["position"])) + return {15: "buy", 20: "exit"}.get(i, "hold") + + n = 30 + flat = {"open": [100.0] * n, "high": [100.5] * n, "low": [99.5] * n, "close": [100.0] * n} + result = Backtester(slippage_pct=0.0, commission_pct=0.0).run(flat, signal) + + assert len(result.trades) == 1 + assert result.trades[0].exit_reason == "signal" and result.trades[0].exit_index == 20 + # Exactly one decision per bar, with the position held at that moment. + assert [i for i, _ in seen] == list(range(15, n)) + assert dict(seen)[16] == 1 and dict(seen)[21] == 0 + + +def test_backtest_retrains_in_memory_agent_but_keeps_loaded_one(tmp_path): + data = _ohlcv(600, seed=11) + rl = RLStrategy({"parameters": dict(FAST)}) + rl.train(data) # trained on everything, including the future test bars + info = rl.before_backtest(data, {}) + assert info["rl_model_source"] == "trained_in_sample" + assert rl.training_summary["train_range"][1] == info["rl_split_index"] + + path = str(tmp_path / "rl.json") + rl.save(path) + loaded = RLStrategy({"parameters": {**FAST, "model_path": path, "load_model": True}}) + q_before = loaded.agent.q.copy() + assert loaded.before_backtest(data, {})["rl_model_source"] == "loaded" + assert np.array_equal(loaded.agent.q, q_before) + + +def test_invalid_set_parameters_leaves_strategy_unchanged(): + rl = RLStrategy({"parameters": dict(FAST)}) + with pytest.raises(ValueError): + rl.set_parameters({"gamma": 2.0}) + assert "gamma" not in rl.parameters + assert rl.rl_config.gamma == RLConfig().gamma diff --git a/ClaudeTrader/utils/backtest.py b/ClaudeTrader/utils/backtest.py index 86b771b..a03a52c 100644 --- a/ClaudeTrader/utils/backtest.py +++ b/ClaudeTrader/utils/backtest.py @@ -35,10 +35,14 @@ logger = logging.getLogger(__name__) -# A signal function receives a window dict (open/high/low/close/volume lists) -# and returns either an action string ('buy'/'sell'/'hold') or an object/dict -# exposing an ``action`` attribute/key. -SignalFunc = Callable[[Dict[str, List[float]]], Any] +# A signal function receives a window dict (open/high/low/close/volume lists +# for bars 0..i plus ``position``: 1 long, -1 short, 0 flat) and returns either +# an action string or an object/dict exposing an ``action`` attribute/key. +# Actions: 'buy' / 'sell' open (or reverse into) a position, 'exit' closes any +# position without opening a new one, and 'hold' changes nothing. +SignalFunc = Callable[[Dict[str, Any]], Any] + +EXIT_ACTIONS = ("exit", "close", "flat") @dataclass @@ -189,93 +193,79 @@ def equity_at(price: float) -> float: # short: cash already includes proceeds; subtract buy-back liability return cash - position["size"] * price + def record_close(exit_index: int, fill: float, reason: str) -> None: + nonlocal cash, position + cash, pnl, ret = self._close_position(cash, position, fill) + trades.append( + Trade( + direction=position["direction"], + entry_index=position["entry_index"], + entry_price=position["entry_price"], + exit_index=exit_index, + exit_price=fill, + size=position["size"], + pnl=pnl, + return_pct=ret, + exit_reason=reason, + ) + ) + trade_pnls.append(pnl) + position = None + for i in range(n): price = close[i] - # --- Manage an open position (check exits) --- + # --- Intrabar stop / target, against the levels in force at the open --- if position is not None: - direction = position["direction"] - # Check stops against the levels that were in force when this - # bar opened; the trailing stop is only ratcheted after the bar - # closes (below) so this bar's close can't tighten its own stop. exit_reason = risk_utils.check_exit( - direction, high[i], low[i], + position["direction"], high[i], low[i], position["stop_loss"], position["take_profit"], ) - - # Close only on an opposite signal. "hold" means "no new - # signal", so the position stays open (SuperTrend, for - # example, emits buy/sell only on the bar the trend flips). - if exit_reason is None and i >= warmup: - action = _extract_action( - signal_func(self._window(open_, high, low, close, volume, i)) - ) - if direction == "long" and action == "sell": - exit_reason = "signal" - elif direction == "short" and action == "buy": - exit_reason = "signal" - - if exit_reason is None and self.trailing_stop and atr_series[i] > 0: - trail_distance = atr_series[i] * self.atr_stop_multiplier - position["stop_loss"] = risk_utils.update_trailing_stop( - direction, price, position["stop_loss"], trail_distance - ) - if exit_reason is not None: fill = self._exit_fill_price( - direction, exit_reason, price, + position["direction"], exit_reason, price, position["stop_loss"], position["take_profit"], open_[i] if has_open else None, ) - cash, pnl, ret = self._close_position(cash, position, fill) - trades.append( - Trade( - direction=direction, - entry_index=position["entry_index"], - entry_price=position["entry_price"], - exit_index=i, - exit_price=fill, - size=position["size"], - pnl=pnl, - return_pct=ret, - exit_reason=exit_reason, - ) - ) - trade_pnls.append(pnl) - position = None + record_close(i, fill, exit_reason) - # --- Look for new entries --- - if position is None and i >= warmup: + # --- One strategy decision per bar, at the close --- + action = "hold" + if i >= warmup: + current = 0 if position is None else (1 if position["direction"] == "long" else -1) action = _extract_action( - signal_func(self._window(open_, high, low, close, volume, i)) + signal_func(self._window(open_, high, low, close, volume, i, current)) ) - atr = atr_series[i] - if atr > 0 and action in ("buy", "sell"): - direction = "long" if action == "buy" else "short" - if not (direction == "short" and not self.allow_short): - cash, position = self._open_position(cash, direction, price, atr, i) + + if position is not None: + direction = position["direction"] + # "hold" keeps the position (SuperTrend, for example, emits + # buy/sell only on the bar the trend flips). An opposite signal + # or an explicit exit closes it. + if action in EXIT_ACTIONS or ( + (direction == "long" and action == "sell") + or (direction == "short" and action == "buy") + ): + record_close(i, self._exit_fill_price(direction, "signal", price, 0.0, 0.0), "signal") + elif self.trailing_stop and atr_series[i] > 0: + # Ratchet only after the bar closes, so a bar's close never + # tightens the stop its own low was tested against. + trail_distance = atr_series[i] * self.atr_stop_multiplier + position["stop_loss"] = risk_utils.update_trailing_stop( + direction, price, position["stop_loss"], trail_distance + ) + + # --- Entries (including a reversal on the bar an opposite signal closed) --- + if position is None and action in ("buy", "sell") and atr_series[i] > 0: + direction = "long" if action == "buy" else "short" + if not (direction == "short" and not self.allow_short): + cash, position = self._open_position(cash, direction, price, atr_series[i], i) equity_curve.append(equity_at(price)) # Force-close any open position at the last close. if position is not None: - fill = self._apply_slippage(position["direction"], "exit", close[-1]) - cash, pnl, ret = self._close_position(cash, position, fill) - trades.append( - Trade( - direction=position["direction"], - entry_index=position["entry_index"], - entry_price=position["entry_price"], - exit_index=n - 1, - exit_price=fill, - size=position["size"], - pnl=pnl, - return_pct=ret, - exit_reason="end_of_data", - ) - ) - trade_pnls.append(pnl) - position = None + record_close(n - 1, self._apply_slippage(position["direction"], "exit", close[-1]), "end_of_data") equity_curve[-1] = cash if not equity_curve: @@ -307,8 +297,10 @@ def _window( close: List[float], volume: List[float], i: int, - ) -> Dict[str, List[float]]: - """Build a data window containing bars 0..i (inclusive).""" + position: int = 0, + ) -> Dict[str, Any]: + """Build a data window containing bars 0..i (inclusive), plus the + current position (1 long, -1 short, 0 flat).""" end = i + 1 return { "open": open_[:end], @@ -316,6 +308,7 @@ def _window( "low": low[:end], "close": close[:end], "volume": volume[:end], + "position": position, } def _open_position( @@ -427,7 +420,8 @@ def backtest_strategy( ohlcv: OHLCV dictionary. config: Optional dict of backtester parameters (commission_pct, slippage_pct, risk_per_trade, atr_period, atr_stop_multiplier, - risk_reward_ratio, trailing_stop, initial_capital, periods_per_year). + risk_reward_ratio, trailing_stop, initial_capital, periods_per_year, + warmup). Returns: A serializable dict describing the backtest outcome. @@ -447,7 +441,13 @@ def backtest_strategy( risk_free_rate=config.get("risk_free_rate", 0.0), ) - result = backtester.run(ohlcv, strategy.generate_signal) + # Strategies that precompute features over the full series (causally) can + # do so once here instead of on every bar. + prepare = getattr(strategy, "prepare", None) + if callable(prepare): + prepare(ohlcv) + + result = backtester.run(ohlcv, strategy.generate_signal, warmup=config.get("warmup")) return result.to_dict() diff --git a/ClaudeTrader/utils/indicators.py b/ClaudeTrader/utils/indicators.py index 3509e72..50e04c1 100644 --- a/ClaudeTrader/utils/indicators.py +++ b/ClaudeTrader/utils/indicators.py @@ -152,6 +152,61 @@ def calculate_atr( return atr +def calculate_supertrend( + high: List[float], + low: List[float], + close: List[float], + atr_period: int = 10, + multiplier: float = 3.0 +) -> Tuple[List[float], List[int], List[float]]: + """Calculate the SuperTrend line, trend direction and ATR. + + All three lists are aligned with ``close`` (length n). ``direction`` is 1 + for an uptrend and -1 for a downtrend. Each value depends only on bars up + to and including its own index. + """ + + n = len(close) + if n == 0: + return [], [], [] + + atr = calculate_atr(high, low, close, atr_period) + + hl2 = [(high[i] + low[i]) / 2 for i in range(n)] + basic_upper = [hl2[i] + multiplier * atr[i] for i in range(n)] + basic_lower = [hl2[i] - multiplier * atr[i] for i in range(n)] + + final_upper = [0.0] * n + final_lower = [0.0] * n + direction = [1] * n + supertrend = [0.0] * n + + final_upper[0] = basic_upper[0] + final_lower[0] = basic_lower[0] + supertrend[0] = final_lower[0] + + for i in range(1, n): + # Carry the band forward unless price/band movement allows a reset. + if basic_upper[i] < final_upper[i - 1] or close[i - 1] > final_upper[i - 1]: + final_upper[i] = basic_upper[i] + else: + final_upper[i] = final_upper[i - 1] + + if basic_lower[i] > final_lower[i - 1] or close[i - 1] < final_lower[i - 1]: + final_lower[i] = basic_lower[i] + else: + final_lower[i] = final_lower[i - 1] + + if direction[i - 1] == 1: + direction[i] = -1 if close[i] < final_lower[i] else 1 + else: + direction[i] = 1 if close[i] > final_upper[i] else -1 + + supertrend[i] = final_lower[i] if direction[i] == 1 else final_upper[i] + + return supertrend, direction, atr + + def calculate_stochastic( high: List[float], low: List[float],