Source code for tradeexecutor.strategy.pandas_trader.runner

"""A strategy runner that executes Trading Strategy Pandas type strategies."""

import datetime
from io import StringIO
from typing import List, Optional
import logging

import pandas as pd

from tradeexecutor.cli.discord import post_logging_discord_image
from tradeexecutor.statistics.in_memory_statistics import refresh_live_strategy_images
from tradeexecutor.strategy.pandas_trader.indicator import IndicatorSet, calculate_and_load_indicators, MemoryIndicatorStorage, call_create_indicators
from tradeexecutor.strategy.pandas_trader.strategy_input import StrategyInput, StrategyInputIndicators
from tradeexecutor.strategy.parameters import StrategyParameters
from tradeexecutor.strategy.pricing_model import PricingModel
from tradeexecutor.strategy.strategy_module import DecideTradesProtocol, DecideTradesProtocol2, DecideTradesProtocol3, DecideTradesProtocol4
from tradeexecutor.strategy.sync_model import SyncModel
from tradeexecutor.strategy.trading_strategy_universe import TradingStrategyUniverse, translate_trading_pair

from tradeexecutor.state.state import State
from import TradeExecution
from tradeexecutor.strategy.runner import StrategyRunner, PreflightCheckFailed
from tradeexecutor.visual.image_output import render_plotly_figure_as_image_file
from tradeexecutor.visual.strategy_state import draw_single_pair_strategy_state, draw_multi_pair_strategy_state

logger = logging.getLogger(__name__)

[docs]class PandasTraderRunner(StrategyRunner): """A trading executor for Pandas math based algorithm."""
[docs] def __init__( self, *args, decide_trades: DecideTradesProtocol | DecideTradesProtocol2 | DecideTradesProtocol3 | DecideTradesProtocol4, max_data_age: datetime.timedelta = None, **kwargs ): super().__init__(*args, **kwargs) self.decide_trades = decide_trades self.max_data_age = max_data_age # Legacy assets sync_model = kwargs.get("sync_model") if sync_model is not None: assert isinstance(sync_model, SyncModel)
def on_data_signal(self): pass
[docs] def on_clock( self, clock: datetime.datetime, strategy_universe: TradingStrategyUniverse, pricing_model: PricingModel, state: State, debug_details: dict, indicators:StrategyInputIndicators | None = None, ) -> List[TradeExecution]: """Run one strategy tick.""" assert isinstance(strategy_universe, TradingStrategyUniverse) universe = strategy_universe.data_universe pd_timestamp = pd.Timestamp(clock) assert state.sync.treasury.last_updated_at is not None, "Cannot do trades before treasury is synced at least once" # All sync models do not emit events correctly yet # assert len(state.sync.treasury.balance_update_refs) > 0, "No deposit detected. Please do at least one deposit before starting the strategy" assert len(strategy_universe.reserve_assets) == 1 # Call the strategy script decide_trades() # callback if self.execution_context.is_version_greater_or_equal_than(0, 5, 0): # DecideTradesProtocolV4 if self.execution_context.mode.is_live_trading(): # Indicators are recalculated for every tick in the live trading assert self.create_indicators is not None, "trading_strategy_engine_version > 0.5, but we lack create_indicators" assert self.parameters is not None, "trading_strategy_engine_version > 0.5, but we lack parameters" indicators = self.calculate_live_indicators(clock, strategy_universe, self.parameters) assert indicators is not None, "indicators not created when running trading_strategy_engine_version=0.5" indicators.prepare_decision_cycle(debug_details["cycle"], pd_timestamp) input = StrategyInput( cycle=debug_details["cycle"], timestamp=pd_timestamp, strategy_universe=strategy_universe, state=state, pricing_model=pricing_model, other_data=debug_details, indicators=indicators, parameters=self.parameters, execution_context=self.execution_context, ) return self.decide_trades( input ) elif self.execution_context.is_version_greater_or_equal_than(0, 4, 0): parameters = self.execution_context.parameters # DecideTradesProtocolV3 parameters["cycle"] = debug_details["cycle"] return self.decide_trades( timestamp=pd_timestamp, parameters=parameters, strategy_universe=strategy_universe, state=state, pricing_model=pricing_model, ) elif self.execution_context.is_version_greater_or_equal_than(0, 3, 0): return self.decide_trades( timestamp=pd_timestamp, strategy_universe=strategy_universe, state=state, pricing_model=pricing_model, cycle_debug_data=debug_details, ) else: return self.decide_trades( timestamp=pd_timestamp, universe=universe, state=state, pricing_model=pricing_model, cycle_debug_data=debug_details, )
[docs] def pretick_check(self, ts: datetime.datetime, universe: TradingStrategyUniverse): """Check the data looks more or less sane.""" assert isinstance(universe, TradingStrategyUniverse) universe = universe.data_universe now_ = ts if len(universe.exchanges) == 0: raise PreflightCheckFailed("Exchange count zero") if universe.pairs.get_count() == 0: raise PreflightCheckFailed("Pair count zero") # Don't assume we have candle or liquidity data e.g. for the testing strategies if universe.candles is not None: if universe.candles.get_candle_count() > 0: start, end = universe.candles.get_timestamp_range() if self.max_data_age is not None: if now_ - end > self.max_data_age: raise PreflightCheckFailed(f"We do not have up-to-date data for candles. Last candles are at {end}")
[docs] def refresh_visualisations(self, state: State, universe: TradingStrategyUniverse): """Updates the visualisation images for the strategy. - Used in Discord (small) - Used on the frontend (large) This is automatically called on trade-executor console startup: docker compose run enzyme-polygon-eth-btc-usdc console To call this manually from the same console with pre-set up runner .. code-block:: shell runner.refresh_visualisations(state, strategy_universe) """ if not self.run_state: # This strategy is not maintaining a run-state # Backtest, simulation, etc."Could not update strategy thinking image data, self.run_state not available") return pair_count = universe.get_pair_count()"Refreshing strategy visualisations: %s, pair count is %d", self.run_state.visualisation, pair_count ) execution_context = self.execution_context if universe.is_empty(): # TODO: Not sure how we end up here"Strategy universe is empty - nothing to report") return try: if pair_count == 1: small_figure = draw_single_pair_strategy_state(state, execution_context, universe, height=512) # Draw the inline plot and expose them tot he web server # TODO: SVGs here are not very readable, have them as a stop gap solution large_figure = draw_single_pair_strategy_state(state, execution_context, universe, height=1024) self.update_strategy_thinking_image_data(small_figure, large_figure) elif 1 < pair_count <= 3: small_figure_combined = draw_multi_pair_strategy_state(state, execution_context, universe, height=1024) large_figure_combined = draw_multi_pair_strategy_state(state, execution_context, universe, height=2048) self.update_strategy_thinking_image_data(small_figure_combined, large_figure_combined) elif 3 < pair_count <=5: small_figure_combined = draw_multi_pair_strategy_state(state, execution_context, universe, height=2048, detached_indicators = False) large_figure_combined = draw_multi_pair_strategy_state(state, execution_context, universe, height=3840, width = 2160, detached_indicators = False) self.update_strategy_thinking_image_data(small_figure_combined, large_figure_combined) else: logger.warning("Charts not yet available for this strategy type. Pair count: %d", pair_count) except Exception as e: # Don't take trade executor down if visualisations fail logger.warning("Could not draw visualisations in refresh_visualisations()") logger.warning("Visualisation exception %s", e, exc_info=e)
[docs] def update_strategy_thinking_image_data(self, small_figure, large_figure): """Update the strategy thinking image data with small, small dark theme, large, and large dark theme images. :param small_image: 512 x 512 image :param large_image: 1920 x 1920 image """ execution_context = self.execution_context refresh_live_strategy_images(self.run_state, execution_context, small_figure, large_figure)
[docs] def report_strategy_thinking( self, strategy_cycle_timestamp: datetime.datetime, cycle: int, universe: TradingStrategyUniverse, state: State, trades: List[TradeExecution], debug_details: dict ): """Strategy admin helpers to understand a live running strategy. - Post latest variables - Draw the single pair strategy visualisation. To manually test the visualisation see: ``. :param strategy_cycle_timestamp: real time lock :param cycle: Cycle number :param universe: Currnet trading universe :param trades: Trades executed on this cycle :param state: Current execution state :param debug_details: Dict of random debug stuff """ # Update charts self.refresh_visualisations(state, universe) visualisation = state.visualisation if universe.is_empty(): # TODO: Not sure how we end up here"Strategy universe is empty - nothing to report") return if universe.is_single_pair_universe(): # Log state buf = StringIO() pair = universe.get_single_pair() candles = universe.data_universe.candles.get_candles_by_pair(pair.internal_id) last_candle = candles.iloc[-1] lag = pd.Timestamp.utcnow().tz_localize(None) - last_candle["timestamp"] print("Strategy thinking", file=buf) print("", file=buf) print(f" Strategy cycle #{cycle}: {strategy_cycle_timestamp} UTC, now is {datetime.datetime.utcnow()}", file=buf) print(f" Last candle at: {last_candle['timestamp']} UTC, market data and action lag: {lag}", file=buf) print(f" Price open:{last_candle['open']} close:{last_candle['close']} {pair.base.token_symbol} / {pair.quote.token_symbol}", file=buf) # Draw indicators for name, plot in visualisation.plots.items(): value = plot.get_last_value() print(f" {name}: {value}", file=buf) small_image = self.run_state.visualisation.small_image_png if small_image is not None: post_logging_discord_image(small_image) else: logger.warning("Chart visualisation missing") else: # Log state buf = StringIO() print("Strategy thinking", file=buf) print(f" Strategy cycle #{cycle}: {strategy_cycle_timestamp} UTC, now is {datetime.datetime.utcnow()}", file=buf) for pair_id, candles in universe.data_universe.candles.get_all_pairs(): pair = universe.data_universe.pairs.get_pair_by_id(pair_id) pair_slug = f"{pair.base_token_symbol} / {pair.quote_token_symbol}" print(f"\n {pair_slug}", file=buf) lag = None timestamp = None last_candle = None if len(candles) > 0: last_candle = candles.iloc[-1] try: timestamp = last_candle["timestamp"] lag = pd.Timestamp.utcnow().tz_localize(None) - timestamp except: logger.warning("Cannot read timestamp") else: logger.warning("Pair %s had not candle data", pair) dex_pair = universe.data_universe.pairs.get_pair_by_id(pair_id) pair = translate_trading_pair(dex_pair) if not pair: logger.warning(f" Pair missing: {dex_pair} - should not happen") else: print(f" Last candle at: {timestamp} UTC, market data and action lag: {lag}", file=buf) if last_candle is not None: print(f" Price open:{last_candle['open']}", file=buf) print(f" Close:{last_candle['close']}") # Draw indicators for name, plot in visualisation.plots.items(): if getattr(plot.pair, "internal_id", None) is None: logger.warning(f" Plot {name} has no pair argument. To see indicator values for individual pairs in a multipair strategy, add pair argument to the `plot_indicator` function in your strategy file.") continue if plot.pair.internal_id != pair_id: continue value = plot.get_last_value() print(f" {name}: {value}", file=buf) # there is already a warning in refresh_visualisations for pair count > 3 if universe.get_pair_count() <= 5: large_image = self.run_state.visualisation.large_image_png post_logging_discord_image(large_image) else:"Strategy visualisation not posted to Discord because pair count of {universe.get_pair_count()} is greater than 5.")
[docs] def calculate_live_indicators( self, timestamp: datetime.datetime, strategy_universe: TradingStrategyUniverse, parameters: StrategyParameters, ) -> StrategyInputIndicators: """Calculate and recalculate indicators in a live trading. - Calculated just before `decide_trades` is called - Recalculated for every cycle :return: Freshly calculated indicators """ # storage = self.indicator_storage"Calculating live indicators for %s", timestamp) storage = MemoryIndicatorStorage(strategy_universe.get_cache_key()) indicators = call_create_indicators( self.create_indicators, parameters, strategy_universe, self.execution_context, timestamp, ) indicator_results = calculate_and_load_indicators( strategy_universe=strategy_universe, storage=storage, execution_context=self.execution_context, indicators=indicators, parameters=self.parameters, timestamp=timestamp, ) strategy_input_indicators = StrategyInputIndicators( strategy_universe, indicator_results=indicator_results, available_indicators=indicators, ) return strategy_input_indicators