-
Notifications
You must be signed in to change notification settings - Fork 0
/
v3.py
178 lines (142 loc) · 6.29 KB
/
v3.py
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 dataclasses import dataclass
from decimal import Decimal
import logging
from typing import Optional, Tuple
from dataclasses_json import DataClassJsonMixin
from uniswap_breakouts.constants.abis import V3_POOL_CONTRACT_ABI
from uniswap_breakouts.uniswap.uniswap_utils import PoolToken, get_pool_token_info
from uniswap_breakouts.utils.web3_utils import contract_call_at_block
logger = logging.getLogger(__name__)
# pylint: disable=too-many-instance-attributes
# dataclass for position reporting doesn't need further grouping
@dataclass(frozen=True)
class V3LiquiditySnapshot(DataClassJsonMixin):
chain: str
block: Optional[int]
token_id: int
current_ratio: Decimal
lower_tick: Decimal
upper_tick: Decimal
token0: PoolToken
num_token0_underlying: Decimal
token1: PoolToken
num_token1_underlying: Decimal
def pool_position_string(chain: str, pool_address: str, nft_id: int, block_no: Optional[int]) -> str:
return f"{chain} - {pool_address} id: {nft_id}" + (
f" at block {block_no}" if block_no is not None else ""
)
def q64_96_to_decimal(q64_96_number: int) -> Decimal:
return Decimal(q64_96_number) / (Decimal(2) ** Decimal(96))
def tick_to_price(tick_index: int) -> Decimal:
return Decimal('1.0001') ** tick_index
def price_outside_below(tick_lower: Decimal, tick_upper: Decimal, liquidity: Decimal) -> Decimal:
return liquidity * (tick_upper.sqrt() - tick_lower.sqrt()) / (tick_lower.sqrt() * tick_upper.sqrt())
def price_outside_above(tick_lower: Decimal, tick_upper: Decimal, liquidity: Decimal) -> Decimal:
return liquidity * (tick_upper.sqrt() - tick_lower.sqrt())
def price_inbetween(
price: Decimal, tick_lower: Decimal, tick_upper: Decimal, liquidity: Decimal
) -> Tuple[Decimal, Decimal]:
token0_position_virtual = (
liquidity * (tick_upper.sqrt() - price.sqrt()) / (price.sqrt() * tick_lower.sqrt())
)
token1_position_virtual = liquidity * (price.sqrt() - tick_lower.sqrt())
return token0_position_virtual, token1_position_virtual
def get_virtual_underlyings_from_range(
ratio: Decimal, lower_ratio: Decimal, upper_ratio: Decimal, liquidity: int
) -> Tuple[Decimal, Decimal]:
if ratio > upper_ratio:
token0_position_virtual = Decimal(0)
token1_position_virtual = price_outside_above(lower_ratio, upper_ratio, Decimal(liquidity))
elif ratio < lower_ratio:
token0_position_virtual = price_outside_below(lower_ratio, upper_ratio, Decimal(liquidity))
token1_position_virtual = Decimal(0)
else:
token0_position_virtual, token1_position_virtual = price_inbetween(
ratio, lower_ratio, upper_ratio, Decimal(liquidity)
)
return token0_position_virtual, token1_position_virtual
def get_price_info_for_pool(chain: str, pool_address: str, block_no: Optional[int]) -> Tuple:
pool_info_result = contract_call_at_block(
chain=chain,
interface_address=pool_address,
implementation_address=pool_address,
fn_name='slot0',
fn_args=[],
block_no=block_no,
abi=V3_POOL_CONTRACT_ABI,
)
return pool_info_result
# pylint: disable=too-many-arguments,too-many-locals
# this is a complex calculation, but I think it's better to keep it all in one function for understanding
def get_underlying_balances(
chain: str,
pool_address: str,
nft_address: str,
nft_impl_address: str,
nft_id: int,
block_no: Optional[int] = None,
) -> V3LiquiditySnapshot:
"""
Get the underlying token balances for a single Uniswap v3 position
see https://atiselsts.github.io/pdfs/uniswap-v3-liquidity-math.pdf.
We use the contract calls to get the necessary inputs into the above formulas for
calculating the underlying positions of the liquidity range
"""
def position_string() -> str:
return pool_position_string(chain, pool_address, nft_id, block_no)
logger.debug("requesting underlying LP balances for V3 position %s", position_string())
token0 = get_pool_token_info(chain, pool_address, 0, V3_POOL_CONTRACT_ABI)
token1 = get_pool_token_info(chain, pool_address, 1, V3_POOL_CONTRACT_ABI)
# the ratio that Uniswap records is a virtual ratio. We will need to adjust by the
# relative decimals of the tokens to get the actual balances later
decimal_adjustment = Decimal(10 ** (token0.decimals - token1.decimals))
logger.debug("getting pool price for %s", position_string())
pool_info_result = get_price_info_for_pool(chain, pool_address, block_no)
sqrt_price_x96 = pool_info_result[0]
price = q64_96_to_decimal(sqrt_price_x96) ** Decimal(2)
logger.info("price of %s for pool %s", price, position_string())
logger.debug("requesting position details for %s", position_string())
positions_info_result = contract_call_at_block(
chain=chain,
interface_address=nft_address,
implementation_address=nft_impl_address,
fn_name='positions',
fn_args=[nft_id],
block_no=block_no,
)
tick_lower = positions_info_result[5]
lower_tick_price = tick_to_price(tick_lower)
tick_upper = positions_info_result[6]
upper_tick_price = tick_to_price(tick_upper)
liquidity = positions_info_result[7]
logger.info(
"position details: lower %s | upper %s | liquidity %s for %s",
lower_tick_price,
upper_tick_price,
liquidity,
position_string(),
)
token0_position_virtual, token1_position_virtual = get_virtual_underlyings_from_range(
price, lower_tick_price, upper_tick_price, liquidity
)
token0_position = token0_position_virtual / (Decimal(10) ** Decimal(token0.decimals))
token1_position = token1_position_virtual / (Decimal(10) ** Decimal(token1.decimals))
logger.info(
"underlying positions of token0 - %s and token1 - %s for %s",
token0_position,
token1_position,
position_string(),
)
snapshot = V3LiquiditySnapshot(
chain=chain,
block=block_no,
token_id=nft_id,
current_ratio=price * decimal_adjustment,
lower_tick=lower_tick_price * decimal_adjustment,
upper_tick=upper_tick_price * decimal_adjustment,
token0=token0,
num_token0_underlying=token0_position,
token1=token1,
num_token1_underlying=token1_position,
)
return snapshot