cyberhybridhub/server/lib/trading/market_data_ingest.dart

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<MarketDataIngestResult> 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<int> _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: <String, dynamic>{
            '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: <String, dynamic>{
                '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: <String, dynamic>{
                'c': prev.close,
                'v': prev.volume,
                't': prev.timestamp.toIso8601String(),
              },
            );
            written++;
          }
        }
      }
    }

    return written;
  }
}