48 feature eval computation

src/pysrc/signal/example_kraken_feature_generator.py

@@ -0,0 +1,38 @@
+from typing import override
+
+from pysrc.adapters.kraken.asset_mappings import asset_to_kraken
+from pysrc.adapters.messages import SnapshotMessage, TradeMessage
+from pysrc.signal.base_feature_generator import BaseFeatureGenerator
+from pysrc.util.types import Asset, Market
+
+
+class ExampleKrakenFeatureGenerator(BaseFeatureGenerator):
+    order_features = ["open", "high", "low", "close"]
+    assets = [Asset.BTC, Asset.ETH, Asset.ADA, Asset.SOL, Asset.DOGE]
+
+    def __init__(self) -> None:
+        pass
+
+    def compute_ohlc(self, trades: list[TradeMessage]) -> list[float]:
+        if not trades:
+            return [0.0, 0.0, 0.0, 0.0]
+        prices = [trade.price for trade in trades]
+        open_price = prices[0]
+        high_price = max(prices)
+        low_price = min(prices)
+        close_price = prices[-1]
+        return [open_price, high_price, low_price, close_price]
+
+    @override
+    def on_tick(
+        self,
+        snapshots: dict[str, SnapshotMessage],
+        trades: dict[str, list[TradeMessage]],
+    ) -> dict[Asset, dict[str, list[float]]]:
+        output = {}
+        for asset in self.assets:
+            asset_key = asset_to_kraken(asset, Market.KRAKEN_SPOT)
+            asset_trades = trades.get(asset_key, [])
+            features = self.compute_ohlc(asset_trades)
+            output[asset] = {"features": features}
+        return output

src/pysrc/test/unit/util/test_feature_eval.py

@@ -0,0 +1,153 @@
+import os
+from datetime import datetime
+
+import pytest
+from pyzstd import CParameter, compress
+
+from pysrc.adapters.messages import SnapshotMessage, TradeMessage
+from pysrc.signal.base_feature_generator import BaseFeatureGenerator
+from pysrc.signal.example_kraken_feature_generator import ExampleKrakenFeatureGenerator
+from pysrc.test.helpers import get_resources_path
+from pysrc.util.feature_eval import Evaluator
+from pysrc.util.types import Asset, Market, OrderSide
+
+resource_path = get_resources_path(__file__)
+
+
+@pytest.fixture
+def client() -> Evaluator:
+    test_features = ["open", "high", "low", "close"]
+    asset = Asset.BTC
+    market = Market.KRAKEN_SPOT
+    start = datetime(year=2024, month=6, day=25)
+    end = datetime(year=2024, month=6, day=26)
+    return Evaluator(
+        features=test_features,
+        asset=asset,
+        market=market,
+        start=start,
+        end=end,
+        resource_path=resource_path,
+    )
+
+
+@pytest.fixture
+def feature_gen() -> BaseFeatureGenerator:
+    return ExampleKrakenFeatureGenerator()
+
+
+def test_feature_calculation(
+    client: Evaluator, feature_gen: BaseFeatureGenerator
+) -> None:
+    snapshots = [
+        SnapshotMessage(
+            time=1719273600,
+            feedcode="XXBTZUSD",
+            market=Market.KRAKEN_SPOT,
+            bids=[],
+            asks=[[1.0, 2.0]],
+        ),
+        SnapshotMessage(
+            time=1719273600,
+            feedcode="XXBTZUSD",
+            market=Market.KRAKEN_SPOT,
+            bids=[[3.0, 4.0], [68717.5, -3000.0]],
+            asks=[[1.0, 2.0], [68717.5, -3000.0]],
+        ),
+        SnapshotMessage(
+            time=1719273601,
+            feedcode="XXBTZUSD",
+            market=Market.KRAKEN_SPOT,
+            bids=[[3.0, 4.0], [68717.5, -3000.0], [7.0, 8.0]],
+            asks=[[1.0, 2.0], [68717.5, -3000.0]],
+        ),
+        SnapshotMessage(
+            time=1719273602,
+            feedcode="XXBTZUSD",
+            market=Market.KRAKEN_SPOT,
+            bids=[[3.0, 4.0], [68717.5, -3000.0], [7.0, 8.0]],
+            asks=[[1.0, 2.0], [68717.5, -3000.0]],
+        ),
+    ]
+
+    snapshots_bytes = b""
+    for s in snapshots:
+        snapshots_bytes += s.to_bytes()
+
+    test_dir_snapshots = resource_path / "snapshots" / "XXBTZUSD"
+    os.makedirs(test_dir_snapshots, exist_ok=True)
+    test_path = test_dir_snapshots / "06_25_2024.bin"
+
+    with open(test_path, "wb") as f:
+        f.write(
+            compress(snapshots_bytes, level_or_option={CParameter.compressionLevel: 10})
+        )
+
+    trades = [
+        TradeMessage(
+            1719273600, "XXBTZUSD", 1, 10.0, 1.0, OrderSide.BID, Market.KRAKEN_SPOT
+        ),
+        TradeMessage(
+            1719273600, "XXBTZUSD", 1, 10.02, 0.5, OrderSide.BID, Market.KRAKEN_SPOT
+        ),
+        TradeMessage(
+            1719273601, "XXBTZUSD", 1, 9.9, 1.5, OrderSide.BID, Market.KRAKEN_SPOT
+        ),
+        TradeMessage(
+            1719273602, "XXBTZUSD", 1, 9.0, 1.5, OrderSide.BID, Market.KRAKEN_SPOT
+        ),
+    ]
+
+    trades_bytes = b""
+    for t in trades:
+        trades_bytes += t.to_bytes()
+
+    test_dir_snapshots = resource_path / "trades" / "XXBTZUSD"
+    os.makedirs(test_dir_snapshots, exist_ok=True)
+    test_path = test_dir_snapshots / "06_25_2024.bin"
+
+    with open(test_path, "wb") as f:
+        f.write(
+            compress(trades_bytes, level_or_option={CParameter.compressionLevel: 10})
+        )
+
+    result = client.calculate_features(feature_gen)
+    assert result["open"][0] == pytest.approx(10.0, rel=1e-7)
+    assert result["open"][1] == pytest.approx(9.9, rel=1e-7)
+    assert result["open"][2] == pytest.approx(9.0, rel=1e-7)
+
+    assert result["high"][0] == pytest.approx(10.02, rel=1e-7)
+    assert result["high"][1] == pytest.approx(9.9, rel=1e-7)
+    assert result["high"][2] == pytest.approx(9.0, rel=1e-7)
+
+    assert result["low"][0] == pytest.approx(10.0, rel=1e-7)
+    assert result["low"][1] == pytest.approx(9.9, rel=1e-7)
+    assert result["low"][2] == pytest.approx(9.0, rel=1e-7)
+
+    assert result["close"][0] == pytest.approx(10.02, rel=1e-7)
+    assert result["close"][1] == pytest.approx(9.9, rel=1e-7)
+    assert result["close"][2] == pytest.approx(9.0, rel=1e-7)
+
+
+def test_feature_evaluation(client: Evaluator) -> None:
+    features = {
+        "open": [1.0, 2.0, 3.0],
+        "high": [2.0, 4.0, 6.0],
+        "low": [-1.0, -2.0, -3.0],
+        "close": [1.0, 2.0, 3.0],
+    }
+
+    target = [4.0, 8.0, 12.0]
+    result = client.evaluate_features(features, target)
+
+    assert result is not None
+
+    expected_result = [
+        [1, 1, -1, 1, 1],
+        [1, 1, -1, 1, 1],
+        [-1, -1, 1, -1, -1],
+        [1, 1, -1, 1, 1],
+        [1, 1, -1, 1, 1],
+    ]
+
+    assert (result == expected_result).any()

src/pysrc/util/feature_eval.py

@@ -0,0 +1,64 @@
+from datetime import datetime
+from pathlib import Path
+
+import numpy as np
+
+from pysrc.adapters.kraken.asset_mappings import asset_to_kraken
+from pysrc.data_loaders.tick_snapshots_data_loader import TickSnapshotsDataLoader
+from pysrc.data_loaders.tick_trades_data_loader import TickTradesDataLoader
+from pysrc.signal.base_feature_generator import BaseFeatureGenerator
+from pysrc.util.types import Asset, Market
+
+
+class Evaluator:
+    def __init__(
+        self,
+        features: list[str],
+        asset: Asset,
+        market: Market,
+        start: datetime,
+        end: datetime,
+        resource_path: Path,
+    ):
+        self._features = features
+        self._asset = asset
+        self._start = start
+        self._end = end
+        self._market = market
+        self._resource_path = resource_path
+
+    def calculate_features(
+        self,
+        generator_client: BaseFeatureGenerator,
+    ) -> dict[str, list[float]]:
+        trades_client = TickTradesDataLoader(
+            self._resource_path, self._asset, self._market, self._start, self._end
+        )
+        snapshots_client = TickSnapshotsDataLoader(
+            self._resource_path, self._asset, self._market, self._start, self._end
+        )
+
+        feature_dict: dict[str, list[float]] = {}
+        for feature in self._features:
+            feature_dict[feature] = []
+
+        asset_str = asset_to_kraken(self._asset, self._market)
+
+        while (trade := trades_client.next()) is not None and (
+            snapshot := snapshots_client.next()
+        ) is not None:
+            calc_features = generator_client.on_tick(
+                {asset_str: snapshot}, {asset_str: trade}
+            )
+            for i in range(len(self._features)):
+                feature = self._features[i]
+                feature_dict[feature].append(calc_features[self._asset]["features"][i])
+        return feature_dict
+
+    def evaluate_features(
+        self, calc_features: dict[str, list[float]], target: list[float]
+    ) -> np.ndarray:
+        input_matrix = []
+        for feature in self._features:
+            input_matrix.append(calc_features[feature])
+        return np.corrcoef(input_matrix, target)