Source code for tradeexecutor.testing.simulated_trader

"""Helper to simulate trades without going through the whole backtesting integration."""

import datetime
from decimal import Decimal

from tradeexecutor.state.state import State, TradeType
from tradeexecutor.state.trade import TradeExecution
from tradeexecutor.state.identifier import TradingPairIdentifier
from tradeexecutor.strategy.pricing_model import PricingModel


[docs]class SimulatedTestTrader: """Create trades. """
[docs] def __init__(self, state: State, pricing_model: PricingModel): self.state = state self.nonce = 1 self.pricing_model = pricing_model
[docs] def buy(self, ts: datetime.datetime, pair: TradingPairIdentifier, amount_in_usd: Decimal) -> TradeExecution: """Buy token (trading pair) for a certain value.""" price_structure = self.pricing_model.get_buy_price(ts, pair, amount_in_usd) price = price_structure.price reserve_currency, exchange_rate = self.state.portfolio.get_default_reserve_asset() position, trade, created = self.state.create_trade( strategy_cycle_at=datetime.datetime.utcnow(), pair=pair, assumed_price=price, quantity=None, reserve=amount_in_usd, trade_type=TradeType.rebalance, reserve_currency=reserve_currency, reserve_currency_price=1.0, pair_fee=pair.fee ) return trade
[docs] def sell(self, pair: TradingPairIdentifier, quantity: Decimal) -> TradeExecution: """Sell token token (trading pair) for a certain quantity.""" reserve_currency, exchange_rate = self.state.portfolio.get_default_reserve_asset() position, trade, created = self.state.create_trade( strategy_cycle_at=datetime.datetime.utcnow(), pair=pair, assumed_price=1.0, quantity=-quantity, reserve=None, trade_type=TradeType.rebalance, reserve_currency=reserve_currency, reserve_currency_price=1.0, pair_fee=pair.fee ) return trade
def simulate_execution(self, state: State, trade: TradeExecution, price_impact=1): ts = trade.opened_at price = trade.planned_price quantity = trade.planned_quantity assert price assert quantity # 2. Capital allocation txid = "0x0" nonce = self.nonce self.state.start_execution(ts, trade, txid, nonce) self.nonce += 1 # 3. Broadcast self.state.mark_broadcasted(ts, trade) # 3. Executed executed_price = price * price_impact if trade.is_buy(): executed_quantity = quantity * Decimal(price_impact) executed_reserve = Decimal(0) else: executed_quantity = quantity executed_reserve = abs(quantity * Decimal(executed_price)) state.mark_trade_success( ts, trade, executed_price, executed_quantity, executed_reserve, lp_fees=0, native_token_price=1, executed_collateral_allocation=trade.planned_collateral_allocation, executed_collateral_consumption=trade.executed_collateral_consumption, )