Use Curve Prices API for pool volume

- Add network.curve_prices
- Add ApiResultError to exceptions
- Add vol_limited_arb.pool_volume.get_pool_volume()
- Compute volume multiplier per asset pair
- Remove volume multiplier modes
- Remove pipelines.utils
- Remove PriceVolume.total_volumes()
- Remove volume from PoolDataCache

changelog.d/20231025_203748_nagakingg_curve_prices_volume.rst

@@ -0,0 +1,16 @@
+Removed
+-------
+- Removed volume multiplier modes
+- Removed pipelines.utils
+- Removed PriceVolume.total_volumes()
+- Removed volume from PoolDataCache
+
+Added
+-----
+- Added network.curve_prices
+- Added ApiResultError to exceptions
+- Added vol_limited_arb.pool_volume.get_pool_volume()
+
+Changed
+-------
+- Volume multipliers now computed individually for each asset pair

curvesim/exceptions/__init__.py

@@ -59,6 +59,10 @@ class NetworkError(CurvesimException):
     """Error for network subpackage."""
 
 
+class ApiResultError(NetworkError):
+    """Raised when API results aren't as expected."""
+
+
 class SimPoolError(CurvesimException):
     """Error in a SimPool operation."""
 

curvesim/iterators/price_samplers/price_volume.py

@@ -89,12 +89,3 @@ def __iter__(self) -> Iterator[PriceVolumeSample]:
             volumes = volumes.to_dict()
 
             yield PriceVolumeSample(price_timestamp, prices, volumes)  # type:ignore
-
-    def total_volumes(self):
-        """
-        Returns
-        -------
-        pandas.Series
-            Total volume for each pairwise coin combination, summed accross timestamps.
-        """
-        return self.volumes.sum().to_dict()

curvesim/network/curve_prices.py

@@ -0,0 +1,126 @@
+"""
+Network connector for Curve Prices API.
+"""
+
+from typing import List
+
+from eth_utils import to_checksum_address
+from pandas import DataFrame, to_datetime
+
+from curvesim.exceptions import ApiResultError, CurvesimValueError
+
+from .http import HTTP
+from .utils import sync
+
+URL = "https://prices.curve.fi/v1/"
+
+CHAIN_ALIASES = {"mainnet": "ethereum"}
+
+
+async def _get_pool_pair_volume(
+    pool_address,
+    base_token_address,
+    quote_token_address,
+    start_ts,
+    end_ts,
+    *,
+    chain="ethereum",
+    interval="day",
+):
+    chain = _chain_from_alias(chain)
+    pool_address = to_checksum_address(pool_address)
+
+    url = URL + f"volume/{chain}/{pool_address}"
+    params = {
+        "main_token": quote_token_address,
+        "reference_token": base_token_address,
+        "start": start_ts,
+        "end": end_ts,
+        "interval": interval,
+    }
+    r = await HTTP.get(url, params=params)
+
+    try:
+        data = r["data"]
+    except KeyError as e:
+        raise ApiResultError(
+            "No historical volume returned for\n"
+            f"Pool: '{pool_address}', Chain: '{chain}',\n"
+            f"Tokens: (base: {base_token_address}, quote: {quote_token_address}),\n"
+            f"Timestamps: (start: {start_ts}, end: {end_ts})"
+        ) from e
+
+    return data
+
+
+async def get_pool_pair_volume(
+    pool_address: str,
+    base_token_address: str,
+    quote_token_address: str,
+    start_ts: int,
+    end_ts: int,
+    *,
+    chain: str = "ethereum",
+    interval: str = "day",
+) -> DataFrame:
+    """
+    Gets historical daily volume for a pair of coins traded in a Curve pool.
+
+    Parameters
+    ----------
+    pool_address: str
+        The Curve pool's address.
+
+    base_token_address: str
+        Address for the base token.
+
+    quote_token_address: str
+        Address for the quote token. Volumes are returned in units of
+
+    start_ts: int
+        Posix timestamp (UTC) for start of query period.
+
+    end_ts: int
+        Posix timestamp (UTC) for end of query period.
+
+    chain: str, default "ethereum"
+        The pool's blockchain (note: currently only "ethereum" supported)
+
+    interval: str, default "day"
+        The sampling interval for the data. Available values: week, day, hour
+
+    Returns
+    -------
+    DataFrame
+        Rows: DateTimeIndex; Columns: volume, fees
+
+    """
+    data: List[dict] = await _get_pool_pair_volume(
+        pool_address,
+        base_token_address,
+        quote_token_address,
+        start_ts,
+        end_ts,
+        chain=chain,
+        interval=interval,
+    )
+
+    df = DataFrame(data, columns=["timestamp", "volume", "fees"])
+    df["timestamp"] = to_datetime(df["timestamp"], unit="s")
+    df.set_index("timestamp", inplace=True)
+    return df
+
+
+def _chain_from_alias(chain):
+    if chain in CHAIN_ALIASES:
+        chain = CHAIN_ALIASES.get(chain, chain)
+
+    if chain != "ethereum":
+        raise CurvesimValueError(
+            "Curve Prices API currently only supports Ethereum chain."
+        )
+
+    return chain
+
+
+get_pool_pair_volume_sync = sync(get_pool_pair_volume)

curvesim/pipelines/vol_limited_arb/__init__.py

@@ -13,7 +13,7 @@
 
 from .. import run_pipeline
 from ..common import DEFAULT_METRICS
-from ..utils import compute_volume_multipliers
+from .pool_volume import get_pool_volume
 from .strategy import VolumeLimitedStrategy
 
 logger = get_logger(__name__)
@@ -31,7 +31,6 @@ def pipeline(
     src="coingecko",
     data_dir="data",
     vol_mult=None,
-    vol_mode=1,
     ncpu=None,
     end=None,
 ):
@@ -83,16 +82,6 @@ def pipeline(
 
             {('DAI', 'USDC'): 0.1, ('DAI', 'USDT'): 0.1, ('USDC', 'USDT'): 0.1}
 
-
-    vol_mode : int, default=1
-        Modes for limiting trade volume.
-
-        1: limits trade volumes proportionally to market volume for each pair
-
-        2: limits trade volumes equally across pairs
-
-        3: mode 2 for trades with meta-pool asset, mode 1 for basepool-only trades
-
     ncpu : int, default=os.cpu_count()
         Number of cores to use.
 
@@ -116,15 +105,10 @@ def pipeline(
     )
 
     if vol_mult is None:
-        total_pool_volume = pool_data_cache.volume
-        total_market_volume = price_sampler.total_volumes()
-        vol_mult = compute_volume_multipliers(
-            total_pool_volume,
-            total_market_volume,
-            pool_metadata.n,
-            pool_metadata.pool_type,
-            mode=vol_mode,
-        )
+        pool_volume = get_pool_volume(pool_metadata, days=days, end=end)
+        vol_mult = pool_volume.sum() / price_sampler.volumes.sum()
+        logger.info("Volume Multipliers:\n%s", vol_mult.to_string())
+        vol_mult = vol_mult.to_dict()
 
     metrics = metrics or DEFAULT_METRICS
     metrics = init_metrics(metrics, pool=pool)