cpy_amm package#

Submodules#

cpy_amm.binance_loader module#

class cpy_amm.binance_loader.BinanceClient(api_key, api_secret)[source]#

Bases: object

create_trade_data(symbol1: str, symbol2: str, pct_volume: float, interval: str, start_time: str, end_time: Optional[str] = None) → cpy_amm.utils.CacheDataFrame[source]#
fetch_klines(symbol, interval, start_time, end_time)[source]#
fetch_klines_parallel(symbol, interval, start_time, end_time, n_intervals=10)[source]#
get_trade_data(symbol: str, interval: str, start_time: str, end_time: Optional[str] = None) → cpy_amm.utils.CacheDataFrame[source]#
cpy_amm.binance_loader.new_binance_client(api_key, secret)[source]#

cpy_amm.gbm_histo module#

cpy_amm.gbm_histo.calibrate_gbm_params(series)[source]#
cpy_amm.gbm_histo.calibrate_mu(prices, dt, n)[source]#
cpy_amm.gbm_histo.calibrate_sigma(prices, dt, n)[source]#

cpy_amm.market module#

class cpy_amm.market.MarketPair(pool_1: cpy_amm.market.Pool, pool_2: cpy_amm.market.Pool, swap_fee: float, mkt_price: float)[source]#

Bases: object

A Market pair managing a liquidity pool made up of reserves of two tokens.

add_liquidity(liq_amount: float, quote_1: cpy_amm.market.MarketQuote, quote_2: cpy_amm.market.MarketQuote)[source]#

Adds a given amount of liquidity in the AMM at the current price of the pool. The amount of token to add in each pool is determined from the market price of the tokens and the current price of the pool.

Args:
liq_amount (float) :

The amount of liquidity to be added expressed in FIAT currency or token eg. USD, EUR, USDT, ETH etc.

quote_1 (MarketQuote) :

The market quote for the token in the first pool

quote_2 (MarketQuote) :

The market quote for the token in the second pool

property asset_base: float#

Current reserve for the base currency expressed in quote currency.

property asset_base_0: float#

Initial reserve for the base currency expressed in quote currency.

assets() → pandas.core.frame.DataFrame[source]#

Calculates assets of the market pair.

Returns:

pd.DataFrame: A DataFrame describing assets of the market pair.

property avg_price: float#

The average execution price for all the trades.

property cp_invariant: float#

The constant product invariant.

describe() → Dict[str, float][source]#

Describes the market pair.

Returns:

Dict[str, float]: A dictionary describing the market pair.

get_delta_reserves() → Tuple[float, float][source]#

The mid price of the trading pair.

get_reserves(trading_pair: str) → Tuple[float, float][source]#

Returns reserves in correct order based on the trading pairs (normal or inversed).

Args:
trading_pair (str) :

The trading pair ticker eg. ETH/USD

Returns:
Tuple[Pool, Pool]:

(Liquidity pool 1, Liquidity pool 2)

Raises:

Exception: If the trading pair is unknown.

property hodl_value: float#

Value of the hold portfolio eg.

current market value of the initial reserves.

property impermanent_loss: float#

The current impermanent loss.

property inverse_ticker: str#

Ticker for the inverse trading pair represented by this market.

property mid_price: float#

The current mid price.

property mid_price_0: float#

The initial mid price.

property mkt_price_ratio: float#

The ratio of the current vs initial market price (moneyness).

perf() → pandas.core.frame.DataFrame[source]#

Calculates the performance of the market pair.

Returns:

pd.DataFrame: A DataFrame describing the performance of the market pair.

property pnl: float#

Cash P&L of the LP position eg.

total fees earned minus impermanent loss.

property roi: float#

ROI of the LP portfolio.

property spread: float#

The spread between the mid price and the market price.

property start_base: float#

Initial reserve for the base currency.

property start_quote: float#

Initial reserve for the quote currency.

property ticker: str#

Ticker for the trading pair represented by this market.

property trade_pnl: float#

Difference of value between the hold portfolio and the LP portfolio.

This is the impermanent loss

property value: float#

Value of the LP portfolio eg.

current market value of the current reserves.

property value_0: float#

Initial value of the hold portfolio eg.

market value of the initial reserves.

class cpy_amm.market.MarketQuote(trading_pair: str, price: float)[source]#

Bases: object

Represents a quote for the market price of a trading pair.

property ticker: str#

Retrieves the ticker for the trading pair.

Returns:

str: The ticker for the trading pair in ‘BASE/QUOTE’ format.

class cpy_amm.market.Pool(ticker: str, initial_deposit: float)[source]#

Bases: object

A class that represents a liquidity pool.

property balance: float#

Retrieves the current balance of the pool.

Returns:

float: The current balance of the pool.

property initial_deposit: float#

Retrieves the initial deposit made to the pool.

Returns:

float: The initial deposit made to the pool.

class cpy_amm.market.TradeOrder(trading_pair: str, order_size: float, transaction_fees: float)[source]#

Bases: object

A trade order for a swap to execute.

classmethod create_default(mkt: cpy_amm.market.MarketPair) → cpy_amm.market.TradeOrder[source]#

Default order equal to 10% of the first pool.

property long: bool#
property short: bool#
cpy_amm.market.new_market(liq_amount: float, quote_1: cpy_amm.market.MarketQuote, quote_2: cpy_amm.market.MarketQuote, swap_fee: float) → cpy_amm.market.MarketPair[source]#

Initializes a market with a given amount of liquidity and market prices for the tokens.

Args:
liq_amount (float) :

The amount of liquidity to be added expressed in FIAT currency or token eg. USD, EUR, USDT, ETH etc.

quote_1 (MarketQuote) :

The market quote for the token in the first pool

quote_2 (MarketQuote) :

The market quote for the token in the second pool

swap_fee (float) :

The transaction fee per swap always paid in the base currency (token in)

Returns:
MarketPair:

New market pair

cpy_amm.market.split_ticker(trading_pair: str) → Tuple[str, str][source]#

Splits a trading pair ticker into base and quote tickers.

Args:

trading_pair (str): The trading pair in the format ‘BASE/QUOTE’.

Returns:

Tuple[str, str]: A tuple of the base and quote tickers.

cpy_amm.market.with_mkt_price(mkt, mkt_price)[source]#

cpy_amm.plotting module#

cpy_amm.plotting.cp_amm_autoviz(mkt: cpy_amm.market.MarketPair, df_trades: Optional[pandas.core.frame.DataFrame] = None, order: Optional[cpy_amm.market.TradeOrder] = None, x_min: Optional[float] = None, x_max: Optional[float] = None, num: Optional[int] = None, precision: Optional[float] = None, plot_width=900, plot_height=600)[source]#

Autoviz for liquidity pools.

Args:
pool_1 (Pool):

Liquidity pool 1

pool_2 (Pool):

Liquidity pool 2

k (float, optional):

Constant product invariant

dx (float, optional):

The order size

x_min (float, optional):

Start of the range for the x-axis

x_max (float, optional):

End of the range for the x-axis

num (int, optional):

Number of points to plot

precision (float, optional):

Precision at which the invariant is evaluated

plot_width (int, optional):

Width of the plot

plot_height (int, optional):

Height of the plot

compact (bool, optional):

If True, 2 plots per row are displayed Else, 1 plot per row is displayed

cpy_amm.plotting.find_cointegrated_pairs(df1, df2)[source]#
cpy_amm.plotting.format_cointegration_result(result)[source]#
cpy_amm.plotting.new_asset_figure(mkt, df_sim, plot_width=900, plot_height=600)[source]#
cpy_amm.plotting.new_constant_product_figure(mkt: cpy_amm.market.MarketPair, x_min: Optional[float] = None, x_max: Optional[float] = None, num: Optional[int] = None, bokeh_figure: Optional[bokeh.plotting.figure.Figure] = None, plot_width=900, plot_height=600)[source]#

Plots the constant product AMM curve Y = K / X

Args:
pool_1 (Pool):

Liquidity pool 1

pool_2 (Pool):

Liquidity pool 2

k (float, optional):

Constant product invariant

x_min (float, optional):

Start of the range for the x-axis

x_max (float, optional):

End of the range for the x-axis

num (int, optional):

Number of points to plot

bokeh_figure (Figure, optional):

Bokeh figure to plot on

plot_width (int, optional):

Width of the plot

plot_height (int, optional):

Height of the plot

cpy_amm.plotting.new_curve_figure(df_sim: pandas.core.frame.DataFrame, title: str = 'No Title', x_label: str = 'No x_label', y_label: str = 'No y_label', y_cols: List[str] = ['No y_cols'], colors: List[str] = ['No colors'], y_percent_format: bool = False, plot_width=900, plot_height=600)[source]#
cpy_amm.plotting.new_df_div(df, plot_width=900, plot_height=600)[source]#
cpy_amm.plotting.new_distrib_figure(df_sim: pandas.core.frame.DataFrame, title: str = 'No Title', x_label: str = 'No x_label', y_label: str = 'No y_label', y_cols: Optional[list] = None, colors: Optional[list] = None, line_dashes: Optional[list] = None, plot_width=900, plot_height=600)[source]#
cpy_amm.plotting.new_fitted_curve_figure(df_sim: pandas.core.frame.DataFrame, title: str = 'No Title', x_label: str = 'No x_label', y_label: str = 'No y_label', x_col: str = 'No x_col', y_cols: List[str] = ['No y_cols'], colors: List[str] = ['No colors'], line_dash: List[str] = ['No line_dash'], y_percent_format: bool = True, plot_width: int = 900, plot_height: int = 600)[source]#
cpy_amm.plotting.new_fitted_pnl_breakdown_figure(df_sim: pandas.core.frame.DataFrame, title: str, x_label: str, y_label: str, x_col: str, y_cols: List[str], colors: List[str], plot_width: int = 900, plot_height: int = 600)[source]#
cpy_amm.plotting.new_fitted_pnl_figure(df_sim: pandas.core.frame.DataFrame, title: str, x_label: str, y_label: str, x_col: str, y_cols: List[str], colors: List[str], line_dash: List[str], plot_width: int = 900, plot_height: int = 600)[source]#
cpy_amm.plotting.new_fitted_price_impact_figure(df_sim: pandas.core.frame.DataFrame, title: str, x_label: str, y_label: str, x_col: str, y_cols: List[str], colors: List[str], line_dash: List[str], plot_width: int = 900, plot_height: int = 600)[source]#
cpy_amm.plotting.new_mkt_price_distrib_figure(df_sim: pandas.core.frame.DataFrame, title: str, x_label: str, y_label: str, y_cols: list, colors: list, line_dashes: list, plot_width=900, plot_height=600)[source]#
cpy_amm.plotting.new_order_book_figure(mkt: cpy_amm.market.MarketPair, x_min: Optional[float] = None, x_max: Optional[float] = None, num: Optional[int] = None, plot_width=900, plot_height=600)[source]#

Plots the cumulative quantity at any mid price according to the formula from the paper “Order Book Depth and Liquidity Provision in Automated Market Makers. Orders under current mid price are on the bid, and those above the ask.

Args:
pool_1 (Pool):

Liquidity pool 1

pool_2 (Pool):

Liquidity pool 2

k (float, optional):

Constant product invariant

x_min (float, optional):

Start of the range for the x-axis

x_max (float, optional):

End of the range for the x-axis

num (int, optional):

Number of points to plot

bokeh_figure (Figure, optional):

Bokeh figure to plot on

plot_width (int, optional):

Width of the plot

plot_height (int, optional):

Height of the plot

cpy_amm.plotting.new_pnl_arb_figure(df_sim: pandas.core.frame.DataFrame, title: str, x_label: str, y_label: str, y_cols: List[str], colors: List[str], y_percent_format: bool, plot_width=900, plot_height=600)[source]#
cpy_amm.plotting.new_pnl_figure(df_sim: pandas.core.frame.DataFrame, title: str, x_label: str, y_label: str, y_cols: List[str], colors: List[str], y_percent_format: bool, plot_width: int = 900, plot_height: int = 600)[source]#
cpy_amm.plotting.new_pool_figure(pool_1: cpy_amm.market.Pool, pool_2: cpy_amm.market.Pool, steps=None, plot_width=900, plot_height=600)[source]#

Plots eveolution of token reserves by steps (time, simulations, blocks etc.).

Args:
pool_1 (Pool):

Liquidity pool 1

pool_2 (Pool):

Liquidity pool 2

steps (list, optional):

List of steps

plot_width (int, optional):

Width of the plot

plot_height (int, optional):

Height of the plot

cpy_amm.plotting.new_portfolio_figure(df_sim: pandas.core.frame.DataFrame, title: str, x_label: str, y_label: str, y_cols: List[str], colors: List[str], y_percent_format: bool, plot_width=900, plot_height=600)[source]#
cpy_amm.plotting.new_price_figure(df_sim: pandas.core.frame.DataFrame, title: str, x_label: str, y_label: str, y_cols: List[str], colors: List[str], y_percent_format: bool, plot_width=900, plot_height=600)[source]#
cpy_amm.plotting.new_price_impact_distrib_figure(df_sim: pandas.core.frame.DataFrame, title: str, x_label: str, y_label: str, y_cols: list, colors: list, line_dashes: list, plot_width=900, plot_height=600)[source]#
cpy_amm.plotting.new_price_impact_figure(mkt: cpy_amm.market.MarketPair, order: Optional[cpy_amm.market.TradeOrder] = None, x_min: Optional[float] = None, x_max: Optional[float] = None, num: Optional[int] = None, precision: Optional[float] = None, bokeh_figure: Optional[bokeh.plotting.figure.Figure] = None, plot_width=900, plot_height=600)[source]#

Plots the constant product AMM curve with price impact range for the oder of size dx.

Args:
pool_1 (Pool):

Liquidity pool 1

pool_2 (Pool):

Liquidity pool 2

k (float, optional):

Constant product invariant

dx (float, optional):

The order size

x_min (float, optional):

Start of the range for the x-axis

x_max (float, optional):

End of the range for the x-axis

num (int, optional):

Number of points to plot

bokeh_figure (Figure, optional):

Bokeh figure to plot on

plot_width (int, optional):

Width of the plot

plot_height (int, optional):

Height of the plot

cpy_amm.plotting.new_roi_distrib_figure(df_sim: pandas.core.frame.DataFrame, title: str, x_label: str, y_label: str, y_cols: list, colors: list, line_dashes: list, plot_width=900, plot_height=600)[source]#
cpy_amm.plotting.new_sim_price_impact_figure(df_sim: pandas.core.frame.DataFrame, title: str, x_label: str, y_label: str, y_cols: List[str], colors: List[str], y_percent_format: bool, plot_width=900, plot_height=600)[source]#
cpy_amm.plotting.new_simulation_figure(mkt, simul, plot_width=900, plot_height=600)[source]#
cpy_amm.plotting.new_trade_figure(df_1, df_2, df_composite, ticker)[source]#
cpy_amm.plotting.plot_asset_reserves(df_sim, pool_1_ticker, pool_2_ticker, steps, plot_width=900, plot_height=600)[source]#
cpy_amm.plotting.plot_price_ratio(df1, df2, symbol1, symbol2)[source]#
cpy_amm.plotting.plot_scatterplot(df1, df2, symbol1, symbol2, n_points=1000)[source]#
cpy_amm.plotting.with_price_info(p, mid: cpy_amm.swap.MidPrice, price_label: str) → bokeh.plotting.figure.figure[source]#

Hover tool with price info for the given point.

cpy_amm.simulation module#

cpy_amm.simulation.resample_df(df: pandas.core.frame.DataFrame, resample_freq: str) → pandas.core.frame.DataFrame[source]#

Resamples the DataFrame based on a given frequency.

Args:

df (pd.DataFrame): The DataFrame to be resampled. resample_freq (str): The resampling frequency.

Returns:

pd.DataFrame: The resampled DataFrame.

cpy_amm.simulation.sim_results(sim_outputs: list) → dict[source]#

Processes simulation outputs to provide a structured result.

Args:

sim_outputs (list): The list of simulation outputs.

Returns:

dict: A dictionary containing the processed simulation results.

cpy_amm.simulation.swap_simulation(mkt: cpy_amm.market.MarketPair, trade_df: pandas.core.frame.DataFrame, strategy: Callable[[dict, cpy_amm.market.MarketPair], List[dict]]) → dict[source]#
cpy_amm.simulation.trade_summary(df_trades: pandas.core.frame.DataFrame) → pandas.core.frame.DataFrame[source]#

Generates a summary of trades.

Args:

df_trades (pd.DataFrame): The DataFrame containing trade data.

Returns:

pd.DataFrame: A DataFrame summarizing the trades.

cpy_amm.swap module#

class cpy_amm.swap.MidPrice(trading_pair: str, x: float, y: float)[source]#

Bases: object

class cpy_amm.swap.PriceImpactRange(start: cpy_amm.swap.MidPrice, mid: cpy_amm.swap.MidPrice, end: cpy_amm.swap.MidPrice)[source]#

Bases: object

cpy_amm.swap.assert_cp_invariant(x: float, y: float, k: float, precision: Optional[float] = None)[source]#

Asserts that the constant product is invariant.

Args:
x (float) :

Reserve of tokens A

y (float) :

Reserve of tokens B

k (float) :

Constant product invariant

precision (float) :

Precision at which the constant product invariant is evaluated

Returns:

None

cpy_amm.swap.calc_arb_trade(mkt: cpy_amm.market.MarketPair, tx_mkt_fee=0.002, tx_network_fee=0) → Tuple[float, float][source]#

Computes the trade size and P&L for an arb trade which will align the mid price of the market pair with the market price.

Args:
mkt (MarketPair) :

The market pair to trade against

tx_mkt_fee (float) :

The market fee to pay for the mkt leg of the arb

tx_network_fee (float) :

The network fee to pay expressed as a percentage of the trade size

Returns:
Tuple[float, float] :

(Trade order size, Trade order P&L)

cpy_amm.swap.constant_product_curve(mkt: cpy_amm.market.MarketPair, x_min: Optional[float] = None, x_max: Optional[float] = None, num: Optional[int] = None) → Tuple[list[float], list[float]][source]#

Computes the AMM curve Y = K/X for a constant product AMM K = XY

Args:
mkt (MarketPair) :

The market pair to trade against

x_min (float) :

minimum value of X

x_max (float) :

maximum value of X

num (float) :

number of points to be computed

Returns:
Tuple[list[float],list[float]] :

(Amount of tokens B out, Swap execution price)

cpy_amm.swap.constant_product_swap(mkt: cpy_amm.market.MarketPair, order: cpy_amm.market.TradeOrder, precision: Optional[float] = None) → Tuple[float, float][source]#

Swap tokens A for tokens B from pool with a XY constant product.

Args:
mkt (MarketPair) :

The market pair to trade against

order (TradeOrder) :

The trade order to execute

precision (float) :

Precision at which constant product invariant is evaluated

Returns:
Tuple[float, float] :

(Amount of tokens B out, Swap execution price)

cpy_amm.swap.mock_constant_product_swap(mkt: cpy_amm.market.MarketPair, order: cpy_amm.market.TradeOrder, precision: Optional[float] = None) → Tuple[float, float][source]#

Swap tokens A for tokens B from pool with a XY constant product.

Args:
mkt (MarketPair) :

The market pair to trade against

order (TradeOrder) :

The trade order to execute

precision (float) :

Precision at which constant product invariant is evaluated

Returns:
Tuple[float, float] :

(Amount of tokens B out, Swap execution price)

cpy_amm.swap.order_book(mkt: cpy_amm.market.MarketPair, x_min: Optional[float] = None, x_max: Optional[float] = None, num: Optional[int] = None)[source]#

Computes the cumulative quantity at any mid price according to the formula from the paper “Order Book Depth and Liquidity Provision in Automated Market Makers”.

Args:
mkt (MarketPair) :

The market pair to trade against

x_min (float) :

minimum value of X

x_max (float) :

maximum value of X

num (float) :

number of points to be computed

Returns:
Tuple[list[float],list[float]] :

(reserves of token A, reserves of token B)

cpy_amm.swap.price_impact_range(mkt: cpy_amm.market.MarketPair, order: Optional[cpy_amm.market.TradeOrder] = None, precision: Optional[float] = None) → cpy_amm.swap.PriceImpactRange[source]#

Price impact of a trade order against a market.

Args:
mkt (MarketPair) :

The market pair to trade against

order (TradeOrder) :

The trade order to execute

precision (float) :

precision at which the invariant is evaluated

Returns:
PriceImpactRange :

Price impact range for given pools and order size

cpy_amm.swap.swap_price(x, y, dx) → float[source]#

Computes the swap execution price for an order size given two pools with reserves x and y.

Args:
x (float) :

Reserve of tokens A

y (float) :

Reserve of tokens B

dx (float) :

Order size

Returns:
float :

Swap execution price

cpy_amm.utils module#

class cpy_amm.utils.CacheDataFrame(obj)[source]#

Bases: pandas.core.frame.DataFrame

cpy_amm.utils.figure_specialization(**metadata)[source]#
cpy_amm.utils.format_df(df, width=None)[source]#
cpy_amm.utils.get_date_format(freq)[source]#
cpy_amm.utils.resample(df, agg)[source]#
cpy_amm.utils.timer_func(func)[source]#

Module contents#

Top-level package for cpy_amm.