strategy.utils.price_volatility
datetime
timedelta
Optional
Tuple
pd
coingecko_clip_window_days
coingecko_dex_get_pool_price
coingecko_dex_get_pool_prices
coingecko_get_price
coingecko_get_prices
format_pair_cex_binance
get_cex_price
get_cex_price_3way_pair
get_cex_prices
get_cex_prices_3way_pair
ModelConfig
Pool
Token
PriceModel
DataFormat
DataSource
ETHNativeChains
convert_time_window
VolatilityModel
COINGECKO_DEX_MISSING_DATA_BUFFER
get_price_and_volatility
def get_price_and_volatility(
price_model: ModelConfig,
volatility_model: ModelConfig,
pool: Pool,
granularity: Optional[str] = "1h",
price_window_multiplier: Optional[float] = 1.0,
volatility_window_multiplier: Optional[float] = 1.0
) -> Tuple[pd.DataFrame, pd.DataFrame]
get_current_price_and_volatility
def get_current_price_and_volatility(
price_model: ModelConfig,
volatility_model: ModelConfig,
pool: Pool,
granularity: Optional[str] = "1h") -> Tuple[int, int]
get_historical_prices
def get_historical_prices(data_source: DataSource,
pool: Pool,
window: int,
granularity: Optional[str] = "1h",
verbose: bool = False) -> pd.DataFrame
Retrieves historical price data for a specified pair of tokens from a DataSource. The granularity and time window for the price data are determined by the strategy's configuration.
Arguments:
data_source
DataSource - The data source to fetch the price data from. (e.g. DataSource.BINANCE)pool
Pool - The pool for which to fetch the price data, containing the Tokens information.window
int - The time window for the price data.granularity
str, optional - The granularity of the price data to fetch. Defaults to "1h".verbose
bool, optional - Whether to print verbose output. Defaults to False.
Returns:
DataFrame
- DF of historical OHLCV for the token pair, adjusted for any necessary inversion.
Raises:
ValueError
- If there is an issue fetching the prices, typically due to network or API errors, or if the specified tokens are not properly recognized by the data source.
Notes:
- The function automatically adjusts for token pairs that are quoted in the opposite direction from the requested pair by inverting the price.
get_current_price
def get_current_price(data_source: DataSource,
pool: Optional[Pool] = None,
base_symbol: Optional[str] = None,
quote_symbol: Optional[str] = None,
chain: Optional[str] = None) -> float