migrating to pydantic

docker-compose.yml

@@ -18,6 +18,8 @@ services:
     ports:
       - 8081:8081
     environment:
+      ME_CONFIG_BASICAUTH_USERNAME: root
+      ME_CONFIG_BASICAUTH_PASSWORD: root
       ME_CONFIG_MONGODB_ADMINUSERNAME: root
       ME_CONFIG_MONGODB_ADMINPASSWORD: root
       ME_CONFIG_MONGODB_URL: mongodb://root:root@mongo:27017/

pyproject.toml

@@ -3,8 +3,9 @@ name = "algorithmic-trader"
 description = "Trading bot with support for realtime trading, backtesting, custom strategies and much more"
 authors = ["Idan Yael"]
 maintainers = ["Idan Yael"]
+packages = [{ include = "algotrader", from = "src" }]
 readme = "README.md"
-version = "0.0.6.post1.dev0+7aabc15"
+version = "0.0.0"
 keywords = ["algo-trader", "trading", "backtesting", "strategy", "bot"]
 license = "MIT"
 classifiers = [
@@ -38,7 +39,7 @@ target-version = ['py311']
 
 
 [tool.poetry-dynamic-versioning]
-enable = false
+enable = true
 
 [tool.poetry.dependencies]
 python = ">=3.10,<3.13"
@@ -52,6 +53,7 @@ coverage = "7.3.2"
 binance-connector = "1.18.0"
 python-dotenv = "0.21.0"
 ibapi = {path = "libs/ib_client"}
+pydantic = "^2.4.2"
 
 [build-system]
 requires = ["poetry-core>=1.0.0", "poetry-dynamic-versioning>=1.0.0,<2.0.0"]
@@ -74,3 +76,23 @@ bug-tracker = "https://github.com/idanya/algo-trader/issues"
 
 [tool.pytest.ini_options]
 pythonpath = ["src"]
+
+[tool.pyright]
+include = ["src/"]
+exclude = ["**/node_modules",
+    "src/algotrader/providers/ib/",
+    "**/__pycache__",
+    "libs/",
+]
+defineConstant = { DEBUG = true }
+venv = "env311"
+
+reportMissingImports = true
+reportMissingTypeStubs = false
+
+pythonVersion = "3.11"
+pythonPlatform = "Linux"
+
+executionEnvironments = [
+  { root = "src" }
+]

src/algotrader/entities/attachments/assets_correlation.py

@@ -0,0 +1,8 @@
+from typing import Literal
+
+from algotrader.entities.attachments.technicals import IndicatorValue
+from algotrader.entities.generic_candle_attachment import GenericCandleAttachment
+
+
+class AssetCorrelation(GenericCandleAttachment[IndicatorValue]):
+    type: Literal["AssetCorrelation"] = "AssetCorrelation"

src/algotrader/entities/attachments/nothing.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import Field
+
+from algotrader.entities.base_dto import BaseEntity
+
+
+class NothingClass(BaseEntity):
+    type: Literal["NothingClass"] = "NothingClass"
+    nothing: str = Field(default="nothing-at-all")

src/algotrader/entities/attachments/returns.py

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from typing import Literal
+
+from algotrader.entities.generic_candle_attachment import GenericCandleAttachment
+
+
+class Returns(GenericCandleAttachment[float]):
+    type: Literal["Returns"] = "Returns"

src/algotrader/entities/attachments/technicals.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+from typing import Literal, Union, List
+
+from algotrader.entities.generic_candle_attachment import GenericCandleAttachment
+
+IndicatorValue = Union[List[float], float]
+
+
+class Indicators(GenericCandleAttachment[IndicatorValue]):
+    type: Literal["Indicators"] = "Indicators"

src/algotrader/entities/attachments/technicals_buckets_matcher.py

@@ -0,0 +1,8 @@
+from typing import Union, List, Literal
+
+from algotrader.entities.bucket import Bucket
+from algotrader.entities.generic_candle_attachment import GenericCandleAttachment
+
+
+class IndicatorsMatchedBuckets(GenericCandleAttachment[Union[List[Bucket], Bucket]]):
+    type: Literal["IndicatorsMatchedBuckets"] = "IndicatorsMatchedBuckets"

src/algotrader/entities/attachments/technicals_normalizer.py

@@ -0,0 +1,10 @@
+from __future__ import annotations
+
+from typing import Literal
+
+from algotrader.entities.attachments.technicals import IndicatorValue
+from algotrader.entities.generic_candle_attachment import GenericCandleAttachment
+
+
+class NormalizedIndicators(GenericCandleAttachment[IndicatorValue]):
+    type: Literal["NormalizedIndicators"] = "NormalizedIndicators"

src/algotrader/entities/base_dto.py

@@ -0,0 +1,11 @@
+from datetime import datetime
+from typing import ClassVar
+
+from pydantic import BaseModel, Field
+
+
+class BaseEntity(BaseModel):
+    _types: ClassVar[dict[str, type]] = {}
+
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_at: datetime = Field(default_factory=datetime.utcnow)

src/algotrader/entities/bucket.py

@@ -1,29 +1,26 @@
 from __future__ import annotations
 
-from typing import Dict, List, Union
+import math
+from typing import List, Union, Optional
 
-from algotrader.entities.serializable import Serializable, Deserializable
+from pydantic import Field
 
+from algotrader.entities.base_dto import BaseEntity
 
-class Bucket(Serializable, Deserializable):
-    def __init__(self, ident: int, start: float = float("-inf"), end: float = float("inf")) -> None:
-        super().__init__()
-        self.ident = ident
-        self.start = start
-        self.end = end
 
-    @classmethod
-    def deserialize(cls, data: Dict) -> Bucket:
-        return Bucket(data["ident"], data["start"], data["end"])
+class Bucket(BaseEntity):
+    ident: float
+    start: Optional[float] = Field(default=-math.inf)
+    end: Optional[float] = Field(default=math.inf)
 
-    def serialize(self) -> Dict:
-        obj = super().serialize()
-        obj.update({"ident": self.ident, "start": self.start, "end": self.end})
+    @property
+    def get_start(self):
+        return self.start or -math.inf
 
-        return obj
+    @property
+    def get_end(self):
+        return self.end or math.inf
 
 
 BucketList = List[Bucket]
 CompoundBucketList = Union[List[BucketList], BucketList]
-
-Bucket(0)

src/algotrader/entities/bucketscontainer.py

@@ -2,15 +2,15 @@
 
 from typing import Dict, Optional, ItemsView
 
+from pydantic import Field
+
+from algotrader.entities.base_dto import BaseEntity
 from algotrader.entities.bucket import Bucket, CompoundBucketList
-from algotrader.entities.serializable import Serializable, Deserializable
 from algotrader.serialization.store import DeserializationService
 
 
-class BucketsContainer(Serializable, Deserializable):
-    def __init__(self) -> None:
-        super().__init__()
-        self.bins: Dict[str, CompoundBucketList] = {}
+class BucketsContainer(BaseEntity):
+    bins: dict[str, CompoundBucketList] = Field(default_factory=dict)
 
     def items(self) -> ItemsView[str, CompoundBucketList]:
         return self.bins.items()

src/algotrader/entities/candle.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import Dict, Optional
+from typing import Optional, Annotated, Literal
 
-from algotrader.entities.candle_attachments import CandleAttachments
-from algotrader.entities.serializable import Serializable, Deserializable
+from pydantic import Field
+
+from algotrader.entities.base_dto import BaseEntity
+from algotrader.entities.candle_attachments import CandleAttachment
 from algotrader.entities.timespan import TimeSpan
-from algotrader.serialization.store import DeserializationService
 
 
 def timestamp_to_str(d: datetime) -> str:
@@ -17,60 +18,26 @@ def str_to_timestamp(d: str) -> datetime:
     return datetime.strptime(d, "%Y%m%d %H:%M:%S.%f")
 
 
-class Candle(Serializable, Deserializable):
-    def __init__(
-        self,
-        symbol: str,
-        time_span: TimeSpan,
-        timestamp: datetime,
-        open: float,
-        close: float,
-        high: float,
-        low: float,
-        volume: float,
-        attachments: Optional[CandleAttachments] = None,
-    ):
-        self.symbol = symbol
-        self.timestamp = timestamp
-        self.time_span = time_span
+class Candle(BaseEntity):
+    type: Literal["Candle"] = "Candle"
+    symbol: str
+    timestamp: datetime
+    time_span: TimeSpan
+
+    open: float
+    close: float
+    high: float
+    low: float
+    volume: float
 
-        self.open = open
-        self.close = close
-        self.high = high
-        self.low = low
-        self.volume = volume
-        self.attachments = attachments or CandleAttachments()
+    attachments: Optional[dict[str, Annotated[CandleAttachment, Field(discriminator="type")]]] = None
 
-    def add_attachement(self, key: str, data: Serializable):
-        self.attachments.add_attachement(key, data)
+    def add_attachment(self, key: str, entity: BaseEntity):
+        if not self.attachments:
+            self.attachments = {}
 
-    def serialize(self) -> Dict:
-        obj = super().serialize()
-        obj.update(
-            {
-                "symbol": self.symbol,
-                "timestamp": timestamp_to_str(self.timestamp),
-                "timespan": self.time_span.value,
-                "open": self.open,
-                "close": self.close,
-                "high": self.high,
-                "low": self.low,
-                "volume": self.volume,
-                "attachments": self.attachments.serialize(),
-            }
-        )
-        return obj
+        self.attachments[key] = entity
 
-    @classmethod
-    def deserialize(cls, data: Dict) -> Candle:
-        return cls(
-            data["symbol"],
-            TimeSpan(data["timespan"]),
-            str_to_timestamp(data["timestamp"]),
-            data["open"],
-            data["close"],
-            data["high"],
-            data["low"],
-            data["volume"],
-            DeserializationService.deserialize(data.get("attachments")),
-        )
+    def get_attachment(self, key: str) -> Optional[CandleAttachment]:
+        if self.attachments:
+            return self.attachments.get(key)

src/algotrader/entities/candle_attachments.py

@@ -1,10 +1,17 @@
 from __future__ import annotations
 
-from typing import Dict, Optional
+from typing import Dict, Optional, Union
 
+from algotrader.entities.attachments.nothing import NothingClass
 from algotrader.entities.serializable import Serializable, Deserializable
+from algotrader.entities.attachments.technicals_buckets_matcher import IndicatorsMatchedBuckets
+from algotrader.entities.attachments.assets_correlation import AssetCorrelation
+from algotrader.entities.attachments.technicals import Indicators
+from algotrader.entities.attachments.technicals_normalizer import NormalizedIndicators
 from algotrader.serialization.store import DeserializationService
 
+CandleAttachment = Union[NothingClass, NormalizedIndicators, Indicators, AssetCorrelation, IndicatorsMatchedBuckets]
+
 
 class CandleAttachments(Serializable, Deserializable):
     def __init__(self) -> None:

src/algotrader/entities/generic_candle_attachment.py

@@ -1,55 +1,34 @@
 from __future__ import annotations
 
-from typing import Dict, TypeVar, Generic, Optional, ItemsView, Union
+from typing import Dict, TypeVar, Generic, Optional, ItemsView
 
-from algotrader.entities.serializable import Serializable, Deserializable
+from pydantic import Field
+
+from algotrader.entities.base_dto import BaseEntity
 
 T = TypeVar("T")
 
 
-class GenericCandleAttachment(Generic[T], Serializable, Deserializable):
-    def __init__(self) -> None:
-        super().__init__()
-        self._data: Dict[str, T] = {}
+class GenericCandleAttachment(Generic[T], BaseEntity):
+    data: Dict[str, T] = Field(default_factory=dict)
 
     def __getitem__(self, key):
-        return self._data[key]
+        return self.data[key]
 
     def set(self, key: str, value: T):
-        self._data[key] = value
+        self.data[key] = value
 
     def get(self, key: str) -> Optional[T]:
-        return self._data[key]
+        return self.data[key]
 
     def items(self) -> ItemsView[str, T]:
         data = {}
-        for k, v in self._data.items():
+        for k, v in self.data.items():
             if k == "__class__":
                 continue
             data.update({k: v})
 
         return data.items()
 
-    @classmethod
-    def deserialize(cls, data: Dict) -> GenericCandleAttachment:
-        obj = GenericCandleAttachment()
-        obj._data = data
-        return obj
-
-    def serialize(self) -> Dict:
-        obj = super().serialize()
-        for k, v in self._data.items():
-            if v:
-                if isinstance(v, list):
-                    obj[k] = [self._serialized_value(x) for x in v]
-                else:
-                    obj[k] = self._serialized_value(v)
-
-        return obj
-
-    @staticmethod
-    def _serialized_value(value: Union[any, Serializable]):
-        return value.serialize() if isinstance(value, Serializable) else value
-
     def has(self, key: str):
-        return key in self._data and self._data[key] is not None
+        return key in self.data and self.data[key] is not None

src/algotrader/pipeline/pipeline.py

@@ -40,7 +40,7 @@ def run(self, context: SharedContext) -> None:
         self.logger.info("Starting pipeline...")
 
         for candle in self.source.read():
-            self.logger.debug("Processing candle: %s\r", candle.serialize())
+            self.logger.debug("Processing candle: %s\r", candle.model_dump())
             self.processor.process(context, candle)
 
         if self.terminator: