Source code for tradeexecutor.ethereum.ethereumtrader

import datetime
from abc import ABC, abstractmethod
from decimal import Decimal
from web3 import Web3
from web3.contract import Contract

from eth_defi.hotwallet import HotWallet
from eth_defi.abi import get_deployed_contract
from tradeexecutor.ethereum.tx import TransactionBuilder

from tradeexecutor.ethereum.uniswap_v2.uniswap_v2_execution import UniswapV2Deployment
from tradeexecutor.state.state import State
from import TradeExecution
from tradeexecutor.state.identifier import TradingPairIdentifier
from tradeexecutor.state.position import TradingPosition

from tradingstrategy.pair import PandasPairUniverse

from tradeexecutor.strategy.routing import RoutingModel

[docs]class EthereumTrader(ABC): """Base class for Uniswap v2 and v3 test traders."""
[docs] def __init__(self, tx_builder: TransactionBuilder, state: State, pair_universe: PandasPairUniverse): assert isinstance(tx_builder, TransactionBuilder) assert isinstance(state, State) assert isinstance(pair_universe, PandasPairUniverse) self.tx_builder = tx_builder self.pair_universe = pair_universe self.state = state self.ts = datetime.datetime(2022, 1, 1, tzinfo=None) self.lp_fees = 2.50 # $2.5 self.gas_units_consumed = 150_000 # 150k gas units per swap self.gas_price = 15 * 10**9 # 15 Gwei/gas unit self.native_token_price = 1 self.confirmation_block_count = 0
@property def web3(self) -> Web3: return self.tx_builder.web3
[docs] @abstractmethod def buy(self, pair: TradingPairIdentifier, amount_in_usd: Decimal, execute=True) -> tuple[TradingPosition, TradeExecution]: """Buy token (trading pair) for a certain value."""
[docs] @abstractmethod def sell(self, pair: TradingPairIdentifier, quantity: Decimal, execute=True) -> tuple[TradingPosition, TradeExecution]: """Sell token token (trading pair) for a certain quantity."""
[docs] @abstractmethod def execute_trades_simple( self, routing_model: RoutingModel, trades: list[TradeExecution], stop_on_execution_failure=True ): """Execute trades on web3 instance. A testing shortcut - Create `BlockchainTransaction` instances - Execute them on Web3 test connection (EthereumTester / Ganache) - Works with single Uniswap test deployment """
# TODO check if duplicated, maybe in routing_model or routing_state
[docs]def get_base_quote_contracts(web3: Web3, pair: TradingPairIdentifier) -> tuple[Contract]: return ( get_mock_erc20_contract(web3, pair.base.address), get_mock_erc20_contract(web3, pair.quote.address), )
[docs]def get_mock_erc20_contract(web3: Web3, address: str): return get_deployed_contract(web3, "ERC20MockDecimals.json", Web3.to_checksum_address(address))