Berack/upo-app-agents
0
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

refactor: remove status attribute from ProductInfo and update related methods to use timestamp_ms

Browse Source
This commit is contained in:
Giacomo Bertolazzi
2025-10-01 23:38:50 +02:00
parent 1c884b67dd
commit 59a38c6e32
13 changed files with 84 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
tests/utils/test_market_aggregator.py
View File

@@ -7,21 +7,30 @@ from app.utils.market_aggregation import aggregate_history_prices, aggregate_pro
@pytest.mark.market
class TestMarketDataAggregator:
def __product(self, symbol: str, price: float, volume: float, status: str, currency: str) -> ProductInfo:
def __product(self, symbol: str, price: float, volume: float, currency: str) -> ProductInfo:
prod = ProductInfo()
prod.id=f"{symbol}-{currency}"
prod.symbol=symbol
prod.price=price
prod.volume_24h=volume
prod.status=status
prod.quote_currency=currency
return prod
def __price(self, timestamp_ms: int, high: float, low: float, open: float, close: float, volume: float) -> Price:
price = Price()
price.timestamp_ms = timestamp_ms
price.high = high
price.low = low
price.open = open
price.close = close
price.volume = volume
return price
def test_aggregate_product_info(self):
products: dict[str, list[ProductInfo]] = {
"Provider1": [self.__product("BTC", 50000.0, 1000.0, "active", "USD")],
"Provider2": [self.__product("BTC", 50100.0, 1100.0, "active", "USD")],
"Provider3": [self.__product("BTC", 49900.0, 900.0, "inactive", "USD")],
"Provider1": [self.__product("BTC", 50000.0, 1000.0, "USD")],
"Provider2": [self.__product("BTC", 50100.0, 1100.0, "USD")],
"Provider3": [self.__product("BTC", 49900.0, 900.0, "USD")],
}
aggregated = aggregate_product_info(products)
@@ -35,18 +44,17 @@ class TestMarketDataAggregator:
avg_weighted_volume = (50000.0 * 1000.0 + 50100.0 * 1100.0 + 49900.0 * 900.0) / (1000.0 + 1100.0 + 900.0)
assert info.volume_24h == pytest.approx(avg_weighted_volume, rel=1e-3)
assert info.status == "active"
assert info.quote_currency == "USD"
def test_aggregate_product_info_multiple_symbols(self):
products = {
"Provider1": [
self.__product("BTC", 50000.0, 1000.0, "active", "USD"),
self.__product("ETH", 4000.0, 2000.0, "active", "USD"),
self.__product("BTC", 50000.0, 1000.0, "USD"),
self.__product("ETH", 4000.0, 2000.0, "USD"),
],
"Provider2": [
self.__product("BTC", 50100.0, 1100.0, "active", "USD"),
self.__product("ETH", 4050.0, 2100.0, "active", "USD"),
self.__product("BTC", 50100.0, 1100.0, "USD"),
self.__product("ETH", 4050.0, 2100.0, "USD"),
],
}
@@ -60,45 +68,33 @@ class TestMarketDataAggregator:
assert btc_info.price == pytest.approx(50050.0, rel=1e-3)
avg_weighted_volume_btc = (50000.0 * 1000.0 + 50100.0 * 1100.0) / (1000.0 + 1100.0)
assert btc_info.volume_24h == pytest.approx(avg_weighted_volume_btc, rel=1e-3)
assert btc_info.status == "active"
assert btc_info.quote_currency == "USD"
assert eth_info is not None
assert eth_info.price == pytest.approx(4025.0, rel=1e-3)
avg_weighted_volume_eth = (4000.0 * 2000.0 + 4050.0 * 2100.0) / (2000.0 + 2100.0)
assert eth_info.volume_24h == pytest.approx(avg_weighted_volume_eth, rel=1e-3)
assert eth_info.status == "active"
assert eth_info.quote_currency == "USD"
def test_aggregate_history_prices(self):
"""Test aggregazione di prezzi storici usando aggregate_history_prices"""
price1 = Price(
timestamp="2024-06-01T00:00:00Z",
price=50000.0,
source="exchange1"
)
price2 = Price(
timestamp="2024-06-01T00:00:00Z",
price=50100.0,
source="exchange2"
)
price3 = Price(
timestamp="2024-06-01T01:00:00Z",
price=50200.0,
source="exchange1"
)
price4 = Price(
timestamp="2024-06-01T01:00:00Z",
price=50300.0,
source="exchange2"
)
prices = {
"Provider1": [
self.__price(1685577600000, 50000.0, 49500.0, 49600.0, 49900.0, 150.0),
self.__price(1685581200000, 50200.0, 49800.0, 50000.0, 50100.0, 200.0),
],
"Provider2": [
self.__price(1685577600000, 50100.0, 49600.0, 49700.0, 50000.0, 180.0),
self.__price(1685581200000, 50300.0, 49900.0, 50100.0, 50200.0, 220.0),
],
}
prices = [price1, price2, price3, price4]
aggregated_prices = aggregate_history_prices(prices)
assert len(aggregated_prices) == 2
assert aggregated_prices[0].timestamp == "2024-06-01T00:00:00Z"
assert aggregated_prices[0].price == pytest.approx(50050.0, rel=1e-3)
assert aggregated_prices[1].timestamp == "2024-06-01T01:00:00Z"
assert aggregated_prices[1].price == pytest.approx(50250.0, rel=1e-3)
aggregated = aggregate_history_prices(prices)
assert len(aggregated) == 2
assert aggregated[0].timestamp_ms == 1685577600000
assert aggregated[0].high == pytest.approx(50050.0, rel=1e-3)
assert aggregated[0].low == pytest.approx(49500.0, rel=1e-3)
assert aggregated[1].timestamp_ms == 1685581200000
assert aggregated[1].high == pytest.approx(50250.0, rel=1e-3)
assert aggregated[1].low == pytest.approx(49800.0, rel=1e-3)