Source code for tradeexecutor.testing.unit_test_trader

"""Simple trade generator without execution."""
import datetime
from decimal import Decimal
from typing import Tuple

import pandas as pd
from tradingstrategy.candle import GroupedCandleUniverse

from tradeexecutor.state.state import State, TradeType
from tradeexecutor.state.position import TradingPosition
from tradeexecutor.state.trade import TradeExecution
from tradeexecutor.state.identifier import TradingPairIdentifier
from tradeexecutor.utils.leverage_calculations import LeverageEstimate


[docs]class UnitTestTrader: """Helper class to generate and settle trades in unit tests. This trade helper is not connected to any blockchain - it just simulates txid and nonce values. """
[docs] def __init__(self, state: State, lp_fees=2.50, price_impact=0.99): self.state = state self.nonce = 1 self.ts = datetime.datetime(2022, 1, 1, tzinfo=None) self.lp_fees = lp_fees self.price_impact = price_impact self.native_token_price = 1
def time_travel(self, timestamp: datetime.datetime): self.ts = timestamp
[docs] def create(self, pair: TradingPairIdentifier, quantity: Decimal, price: float) -> Tuple[TradingPosition, TradeExecution]: """Open a new trade.""" # 1. Plan position, trade, created = self.state.create_trade( strategy_cycle_at=self.ts, pair=pair, quantity=quantity, reserve=None, assumed_price=price, trade_type=TradeType.rebalance, reserve_currency=pair.quote, reserve_currency_price=1.0, planned_mid_price=price, pair_fee=pair.fee ) self.ts += datetime.timedelta(seconds=1) return position, trade
def create_and_execute(self, pair: TradingPairIdentifier, quantity: Decimal, price: float, underflow_check=True) -> Tuple[TradingPosition, TradeExecution]: assert price > 0 assert quantity != 0 price_impact = self.price_impact # 1. Plan position, trade = self.create( pair=pair, quantity=quantity, price=price) # 2. Capital allocation txid = hex(self.nonce) nonce = self.nonce self.state.start_execution(self.ts, trade, txid, nonce, underflow_check=underflow_check) # 3. broadcast self.nonce += 1 self.ts += datetime.timedelta(seconds=1) self.state.mark_broadcasted(self.ts, trade) self.ts += datetime.timedelta(seconds=1) # 4. 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)) self.state.mark_trade_success( self.ts, trade, executed_price, executed_quantity, executed_reserve, self.lp_fees, self.native_token_price) return position, trade
[docs] def set_perfectly_executed(self, trade: TradeExecution): """Sets trade to a executed state. - There are no checks whether the wallet contains relevant balances or not """ # 2. Capital allocation txid = hex(self.nonce) nonce = self.nonce self.state.start_execution(self.ts, trade, txid, nonce, underflow_check=False) # 3. broadcast self.nonce += 1 self.ts += datetime.timedelta(seconds=1) self.state.mark_broadcasted(self.ts, trade) self.ts += datetime.timedelta(seconds=1) # 4. executed executed_price = trade.planned_price executed_collateral_consumption = trade.planned_collateral_consumption executed_collateral_allocation = trade.planned_collateral_allocation if trade.is_spot(): if trade.is_buy(): executed_quantity = trade.planned_quantity executed_reserve = Decimal(0) # TODO: Check if we can reorg code here more cleanly else: # short reduction also changes the reserve (releases collateral) executed_quantity = trade.planned_quantity executed_reserve = trade.planned_reserve else: executed_quantity = trade.planned_quantity executed_reserve = trade.planned_reserve if trade.planned_loan_update: trade.executed_loan_update = trade.planned_loan_update lp_fees = trade.lp_fees_estimated or self.lp_fees self.state.mark_trade_success( self.ts, trade, executed_price, executed_quantity, executed_reserve, lp_fees, self.native_token_price, executed_collateral_consumption=executed_collateral_consumption, executed_collateral_allocation=executed_collateral_allocation, )
def buy(self, pair, quantity, price) -> Tuple[TradingPosition, TradeExecution]: return self.create_and_execute(pair, quantity, price) def prepare_buy(self, pair, quantity, price) -> Tuple[TradingPosition, TradeExecution]: return self.create(pair, quantity, price) def sell(self, pair, quantity, price) -> Tuple[TradingPosition, TradeExecution]: return self.create_and_execute(pair, -quantity, price) def buy_with_price_data(self, pair, quantity, candle_universe: GroupedCandleUniverse) -> Tuple[TradingPosition, TradeExecution]: price = candle_universe.get_closest_price(pair.internal_id, pd.Timestamp(self.ts)) return self.create_and_execute(pair, quantity, float(price)) def sell_with_price_data(self, pair, quantity, candle_universe: GroupedCandleUniverse) -> Tuple[TradingPosition, TradeExecution]: price = candle_universe.get_closest_price(pair.internal_id, pd.Timestamp(self.ts)) return self.create_and_execute(pair, -quantity, float(price)) def open_short(self, pair, quantity, price, leverage=1) -> tuple[TradingPosition, TradeExecution]: assert pair.kind.is_leverage() estimation = LeverageEstimate.open_short( starting_reserve=quantity, leverage=leverage, borrowed_asset_price=price, shorting_pair=pair, fee=pair.get_pricing_pair().fee, ) position, trade, _ = self.state.trade_short( strategy_cycle_at=self.ts, pair=pair, borrowed_quantity=-estimation.borrowed_quantity, collateral_quantity=quantity, borrowed_asset_price=price, trade_type=TradeType.rebalance, reserve_currency=pair.get_pricing_pair().quote, collateral_asset_price=1.0, planned_collateral_consumption=estimation.additional_collateral_quantity, lp_fees_estimated=estimation.lp_fees, ) self.ts += datetime.timedelta(seconds=1) return position, trade