"""Reverse engineering Trading Strategy trading universe from the local EVM tester Uniswap v2 deployment."""
from typing import List, Optional
import pandas as pd
from web3 import Web3
from eth_defi.uniswap_v2.deployment import UniswapV2Deployment
from eth_defi.uniswap_v3.deployment import UniswapV3Deployment
from tradeexecutor.state.identifier import TradingPairIdentifier
from tradingstrategy.chain import ChainId
from tradingstrategy.exchange import ExchangeUniverse, Exchange, ExchangeType
from tradingstrategy.pair import DEXPair, PandasPairUniverse
[docs]def create_pair_universe(web3: Web3, exchange: Exchange, pairs: List[TradingPairIdentifier]) -> PandasPairUniverse:
"""Creates a PairUniverse from on-chain test data.
A test trading universe.
:param web3:
Web3 connection to the test backend
:param exchange:
Exchange under which pairs are assigned
:param pairs:
Trading pairs that will form this pair universe
"""
chain_id = ChainId(web3.eth.chain_id)
data = []
for p in pairs:
assert p.exchange_address
assert p.base.decimals
assert p.quote.decimals
assert p.base.address != p.quote.address
assert p.fee is not None, f"Pair missing fee {p}"
# pair_id cannot be None here since it is enforced in the TradingPairIdentifier hash method
# so if no internal_id is already set, we use the integer version of the identifier as the pair_id
dex_pair = DEXPair(
pair_id=p.internal_id or int(p.get_identifier(), 16),
chain_id=chain_id,
exchange_id=exchange.exchange_id if exchange else 1,
exchange_address=p.exchange_address if exchange else None,
address=p.pool_address,
dex_type=ExchangeType.uniswap_v2,
base_token_symbol=p.base.token_symbol,
quote_token_symbol=p.quote.token_symbol,
token0_symbol=p.base.token_symbol,
token1_symbol=p.quote.token_symbol,
token0_address=p.base.address,
token1_address=p.quote.address,
token0_decimals=p.base.decimals,
token1_decimals=p.quote.decimals,
fee=int(p.fee * 10000),
)
data.append(dex_pair.to_dict())
df = pd.DataFrame(data)
# Need exchange universe in order to use `get_pair`
if exchange:
exchange_universe = ExchangeUniverse.from_collection([exchange])
else:
exchange_universe = None
return PandasPairUniverse(df, exchange_universe=exchange_universe)
[docs]def create_exchange_universe(web3: Web3, uniswaps: List[UniswapV2Deployment | UniswapV3Deployment]) -> ExchangeUniverse:
"""Create an exchange universe with a list of Uniswap v2 deployments."""
exchanges = {}
chain_id = ChainId(web3.eth.chain_id)
for u in uniswaps:
e = Exchange(
chain_id=chain_id,
chain_slug="tester",
exchange_id=int(u.factory.address, 16),
exchange_slug="uniswap_tester",
address=u.factory.address,
exchange_type=ExchangeType.uniswap_v2,
pair_count=99999,
)
exchanges[e.exchange_id] = e
return ExchangeUniverse(exchanges=exchanges)