-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
178 lines (138 loc) · 6.82 KB
/
Copy pathmain.py
File metadata and controls
178 lines (138 loc) · 6.82 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
from starkbiter import *
import typing as t
import os
import asyncio
ALCHEMY_API_KEY = os.getenv("ALCHEMY_KEY")
ARGENT_ACCOUNT_CLASS_HASH = "0x36078334509b514626504edc9fb252328d1a240e4e948bef8d0c08dff45927f"
EKUBO_CORE_ADDRESS = "0x00000005dd3d2f4429af886cd1a3b08289dbcea99a294197e9eb43b0e0325b4b" # MAINNET
class TraderAgent:
def __init__(self, environment: Environment):
self.token0 = "ETH"
self.token1 = "USDC"
self.environment: Environment = environment
self.middleware: t.Optional[Middleware] = None
self.account: t.Optional[Account] = None
self.token0_data: t.Optional[BridgedToken] = None
self.token1_data: t.Optional[BridgedToken] = None
self.swapper_address: t.Optional[str] = None
async def prepare(self):
self.middleware = await self.environment.create_middleware()
self.account = await self.middleware.create_account(ARGENT_ACCOUNT_CLASS_HASH)
await self.middleware.top_up_balance(self.account.address, 1000000000000000000, Tokens.STRK)
self.token0_data = await self.environment.get_token_data(self.token0)
self.token1_data = await self.environment.get_token_data(self.token1)
# TODO(baitcode): validate token0 address < token1 address
# Deploying swapper contract.
# 1. Declaring swapper
swapper_class_hash = await self.middleware.declare_contract(known_calls.ekubo.swapper.SWAPPER_CONTRACT_SIERRA)
# 2. UDC deploy call using traders account (must be the money)
res = await self.account.execute(calls=[
known_calls.udc.deploy(
class_hash=swapper_class_hash,
constructor_calldata=[
# This is ekubo core address. Swapper will send swaps there.
int(EKUBO_CORE_ADDRESS, 16),
# This is withdrawal address. All funds are withdrawn as Swap ends.
# This might be undesirable, but for now we use it.
int(self.account.address, 16),
]
)
])
# Getting the deployed contract address. (By default UDC does not return it)
self.swapper_address = await self.middleware.get_deployed_contract_address(res)
# Let's send some moneys to swapper, as swapper uses it's balance to perform swap agains EKUBO
await self.middleware.top_up_balance(self.swapper_address, 1000000000000000000, Tokens.ETH)
print("Swapper deployed:", res)
# Lets calculate sqrt_ratio_limit for swaps.
# The API looks awful. I was in a rush, I'm open to suggestions.
get_pool_price_call, parse_res = known_calls.ekubo.core.get_pool_price_call(
EKUBO_CORE_ADDRESS, known_calls.ekubo.PoolKey(
token0=self.token0_data.l2_token_address,
token1=self.token1_data.l2_token_address,
fee_tick=known_calls.ekubo.FeeAndTickSpacing.FEE_0_01_PRECISION_0_02,
)
)
sqrt_ratio_res = parse_res(await self.account.call(get_pool_price_call))
print("Sqrt ratio", sqrt_ratio_res)
# # Let's get liquidity.
get_pool_liquidity_call, parse_res = known_calls.ekubo.core.get_pool_liquidity_call(
EKUBO_CORE_ADDRESS, known_calls.ekubo.PoolKey(
token0=self.token0_data.l2_token_address,
token1=self.token1_data.l2_token_address,
fee_tick=known_calls.ekubo.FeeAndTickSpacing.FEE_0_01_PRECISION_0_02,
)
)
liquidity_res = parse_res(await self.account.call(get_pool_liquidity_call))
print("Liquidity", liquidity_res)
return
async def act(self):
print("Trader agent acting...")
res = await self.middleware.get_balance(self.swapper_address, Tokens.ETH)
print("Trader agent ETH balance:", res)
res = await self.account.execute(calls=[
known_calls.ekubo.swapper.swap(
swapper_address=self.swapper_address,
key=known_calls.ekubo.PoolKey(
token0=self.token0_data.l2_token_address,
token1=self.token1_data.l2_token_address,
fee_tick=known_calls.ekubo.FeeAndTickSpacing.FEE_5_PRECISION_10,
),
amount=1000000000000000, # 0.001 ETH
# This is the minimum sqrt ratio limit
sqrt_ratio_limit=known_calls.ekubo.SQRT_RATE_LIMIT_MIN,
token=self.token0_data.l2_token_address,
)
])
print("Swap result:", res)
await self.middleware.get_balance(self.swapper_address, Tokens.ETH)
print("Trader agent ETH balance:", res)
fork_block = 1521205 # 1593123
fork_hash = "0x7aabf76192d3d16fe8bda54c0e7d0a9843c21fe20dd23704366bad38d57dc30"
node_url = f"https://starknet-mainnet.g.alchemy.com/starknet/version/rpc/v0_8/{ALCHEMY_API_KEY}/"
fork = ForkParams(
# IMPORTANT: Do not use blast API, it's API format is incompatible with starknet-rs lib
# url=node_url,
url=f"sqlite://{os.getcwd()}/mainnet.sqlite",
# This is the block number to fork from
block_number=fork_block,
# This is the block hash to fork from. Yes we need both parameters for environment to work.
# This can be optimised, but I didn't want to spend time on it.
block_hash=fork_hash
)
async def simulation(starting_block_number):
# This spins the environment up
async with get_environment(chain=Chains.MAINNET, fork=fork) as env:
# This is a separate middleware I use to set environment up.
middleware = await env.create_middleware()
await middleware.set_gas_price(ALL_PRICES_1)
trader = TraderAgent(env)
await trader.prepare()
iteration = 0
events = []
while True:
print("Events:", events)
await trader.act()
# await arbitrager.act()
starting_block_number += 1
filter_ekubo_swaps = EventFilter(
from_address=EKUBO_CORE_ADDRESS,
event_name=b"Swapped",
)
# Will only apply transactions that emit ekubo Swapped events from core contract on the mainnet. Will not create a block.
await middleware.replay_block_with_txs(
node_url,
BlockNumber(starting_block_number),
filters=[filter_ekubo_swaps]
)
block_hash = await middleware.create_block()
print("Block hash:", block_hash)
events = await middleware.get_block_events(
from_block=BlockHash(block_hash),
keys=filter_ekubo_swaps.keys,
from_address=filter_ekubo_swaps.from_address
)
iteration += 1
if iteration > 5:
print("Simulation completed after 5 iterations.")
break
asyncio.run(simulation(fork_block))