import '../alpaca/alpaca_market_data_client.dart'; import '../alpaca/alpaca_models.dart'; import 'market_data_db.dart'; import 'trading_config.dart'; import 'user_trading_state_db.dart'; /// Result of one [MarketDataIngest.runIfDue] cycle. class MarketDataIngestResult { MarketDataIngestResult({ required this.snapshotsWritten, required this.inputsFetched, required this.inputsSkipped, required this.httpRequests, }); final int snapshotsWritten; final int inputsFetched; final int inputsSkipped; final int httpRequests; } /// Fetches Alpaca market data per [DataInputConfig] and writes snapshots. class MarketDataIngest { MarketDataIngest({ required MarketDataDb marketDataDb, required UserTradingStateDb tradingStateDb, required AlpacaMarketDataClient alpacaClient, }) : _marketDataDb = marketDataDb, _tradingStateDb = tradingStateDb, _alpacaClient = alpacaClient; final MarketDataDb _marketDataDb; final UserTradingStateDb _tradingStateDb; final AlpacaMarketDataClient _alpacaClient; int _httpRequests = 0; /// Exposed for tests (mock HTTP call count). int get httpRequestCount => _httpRequests; /// Runs ingest for each [config.dataInputs] entry when poll interval has elapsed. Future runIfDue({ required String firebaseUid, required EffectiveTradingConfig config, DateTime? now, }) async { _httpRequests = 0; final DateTime tick = (now ?? DateTime.now()).toUtc(); int snapshotsWritten = 0; int inputsFetched = 0; int inputsSkipped = 0; for (final DataInputConfig input in config.dataInputs) { final DateTime? lastFetch = await _tradingStateDb.getInputLastFetch(firebaseUid, input.id); if (lastFetch != null) { final Duration elapsed = tick.difference(lastFetch); if (elapsed.inSeconds < input.pollIntervalSeconds) { inputsSkipped++; continue; } } snapshotsWritten += await _ingestDataInput(input); await _tradingStateDb.recordInputFetch(firebaseUid, input.id, tick); inputsFetched++; } return MarketDataIngestResult( snapshotsWritten: snapshotsWritten, inputsFetched: inputsFetched, inputsSkipped: inputsSkipped, httpRequests: _httpRequests, ); } Future _ingestDataInput(DataInputConfig input) async { int written = 0; final bool needsBars = input.metrics.contains('daily_bar') || input.metrics.contains('prev_close'); final bool needsTrade = input.metrics.contains('last_trade'); AlpacaBarsResponse? barsResponse; if (needsBars) { barsResponse = await _alpacaClient.getDailyBars(input.symbols, limit: 2); _httpRequests++; } if (needsTrade) { for (final String symbol in input.symbols) { final AlpacaLatestTradeResponse latest = await _alpacaClient.getLatestTrade(symbol); _httpRequests++; await _marketDataDb.upsertSnapshot( symbol: symbol, assetClass: input.assetClass, feed: input.feed, metric: 'last_trade', timeframe: 'tick', price: latest.trade.price, volume: latest.trade.size, asOf: latest.trade.timestamp, raw: { 'p': latest.trade.price, 's': latest.trade.size, 't': latest.trade.timestamp.toIso8601String(), }, ); written++; } } if (barsResponse != null) { for (final String symbol in input.symbols) { if (input.metrics.contains('daily_bar')) { final AlpacaBar? bar = barsResponse.latestBar(symbol); if (bar != null) { await _marketDataDb.upsertSnapshot( symbol: symbol, assetClass: input.assetClass, feed: input.feed, metric: 'daily_bar', timeframe: 'tick', price: bar.close, volume: bar.volume, asOf: bar.timestamp, raw: { 'c': bar.close, 'v': bar.volume, 't': bar.timestamp.toIso8601String(), }, ); written++; } } if (input.metrics.contains('prev_close')) { final AlpacaBar? prev = barsResponse.previousDailyBar(symbol); if (prev != null) { await _marketDataDb.upsertSnapshot( symbol: symbol, assetClass: input.assetClass, feed: input.feed, metric: 'prev_close', timeframe: 'tick', price: prev.close, volume: prev.volume, asOf: prev.timestamp, raw: { 'c': prev.close, 'v': prev.volume, 't': prev.timestamp.toIso8601String(), }, ); written++; } } } } return written; } }