diff options
| author | TheSiahxyz <164138827+TheSiahxyz@users.noreply.github.com> | 2026-04-01 15:56:35 +0900 |
|---|---|---|
| committer | TheSiahxyz <164138827+TheSiahxyz@users.noreply.github.com> | 2026-04-01 15:56:35 +0900 |
| commit | 33b14aaa2344b0fd95d1629627c3d135b24ae102 (patch) | |
| tree | 90b214758bc3b076baa7711226a1a1be6268e72e /services/data-collector/src/data_collector/storage.py | |
| parent | 9360f1a800aa29b40399a2f3bfbfcf215a04e279 (diff) | |
feat: initial trading platform implementation
Binance spot crypto trading platform with microservices architecture:
- shared: Pydantic models, Redis Streams broker, asyncpg DB layer
- data-collector: Binance WebSocket/REST market data collection
- strategy-engine: Plugin-based strategy execution (RSI, Grid)
- order-executor: Order execution with risk management
- portfolio-manager: Position tracking and PnL calculation
- backtester: Historical strategy testing with simulator
- cli: Click-based CLI for all operations
- Docker Compose orchestration with Redis and PostgreSQL
- 24 test files covering all modules
Diffstat (limited to 'services/data-collector/src/data_collector/storage.py')
| -rw-r--r-- | services/data-collector/src/data_collector/storage.py | 24 |
1 files changed, 24 insertions, 0 deletions
diff --git a/services/data-collector/src/data_collector/storage.py b/services/data-collector/src/data_collector/storage.py new file mode 100644 index 0000000..1e40b82 --- /dev/null +++ b/services/data-collector/src/data_collector/storage.py @@ -0,0 +1,24 @@ +"""Candle storage: persists to DB and publishes to Redis.""" +from shared.events import CandleEvent +from shared.models import Candle + + +class CandleStorage: + """Stores candles in the database and publishes CandleEvents to Redis.""" + + def __init__(self, db, broker) -> None: + self._db = db + self._broker = broker + + async def store(self, candle: Candle) -> None: + """Insert candle into DB and publish a CandleEvent to the Redis stream.""" + await self._db.insert_candle(candle) + + event = CandleEvent(data=candle) + stream = f"candles.{candle.symbol}" + await self._broker.publish(stream, event.to_dict()) + + async def store_batch(self, candles: list[Candle]) -> None: + """Store multiple candles one by one.""" + for candle in candles: + await self.store(candle) |
