stock-bot/apps/stock/data-ingestion/src/handlers/qm/actions/prices.action.ts

/**
 * QM Prices Actions - Fetch and update daily price data
 */

import type { ExecutionContext } from '@stock-bot/handlers';
import type { QMHandler } from '../qm.handler';
import { QM_CONFIG, QM_SESSION_IDS } from '../shared/config';
import { QMOperationTracker } from '../shared/operation-tracker';
import { QMSessionManager } from '../shared/session-manager';

// Cache tracker instance
let operationTracker: QMOperationTracker | null = null;

/**
 * Get or initialize the operation tracker
 */
async function getOperationTracker(handler: QMHandler): Promise<QMOperationTracker> {
  if (!operationTracker) {
    const { initializeQMOperations } = await import('../shared/operation-registry');
    operationTracker = await initializeQMOperations(handler.mongodb, handler.logger);
  }
  return operationTracker;
}

/**
 * Update daily prices for a single symbol
 */
export async function updatePrices(
  this: QMHandler,
  input: {
    qmSearchCode: string;
    lastRecordDate?: Date;
  },
  _context?: ExecutionContext
): Promise<{
  success: boolean;
  qmSearchCode: string;
  message: string;
  data?: any;
}> {
  const { qmSearchCode, lastRecordDate } = input;
  
  this.logger.info(`Fetching daily prices ${qmSearchCode}`, { qmSearchCode });

  const sessionManager = QMSessionManager.getInstance();
  sessionManager.initialize(this.cache, this.logger);

  // Get a session - you'll need to add the appropriate session ID for prices
  const sessionId = QM_SESSION_IDS.PRICES;
  const session = await sessionManager.getSession(sessionId);
  
  if (!session || !session.uuid) {
    throw new Error(`No active session found for QM prices`);
  }

  try {
    // Build API request for daily prices
    const searchParams = new URLSearchParams({
    zeroTradeDays: 'false',
    start: lastRecordDate?.toISOString().split('T')[0] ?? '1960-01-01',
    interval: '1',
    marketSession: 'mkt',
    freq: 'day',
    adjusted: 'false',
    adjustmentType: 'none',
    unadjusted: 'true',
    datatype: 'eod',
    symbol: qmSearchCode,
    });
    // https://app.quotemedia.com/datatool/getEnhancedChartData.json?zeroTradeDays=false&start=2025-06-22&interval=1&marketSession=mkt&freq=day&adjusted=true&adjustmentType=none&unadjusted=false&datatype=int&symbol=AAPL
    // TODO: Update with correct prices endpoint
    const apiUrl = `${QM_CONFIG.PRICES_URL}?${searchParams.toString()}`;

    const response = await fetch(apiUrl, {
      method: 'GET',
      headers: session.headers,
      proxy: session.proxy,
    });

    if (!response.ok) {
      throw new Error(`QM API request failed: ${response.status} ${response.statusText}`);
    }

    const responseData = await response.json();

    // Update session success stats
    await sessionManager.incrementSuccessfulCalls(sessionId, session.uuid);

    const priceData = responseData.results?.history[0].eoddata || [];
    
    if(!priceData || priceData.length === 0) {
      this.logger.warn(`No price data found for symbol ${qmSearchCode}`);
      return {
        success: false,
        qmSearchCode,
        message: `No price data found for symbol ${qmSearchCode}`
      };
    }

    // Process and store price data
    if (priceData && priceData.length > 0) {
      // Store prices in a separate collection
      const processedPrices = priceData.map((price: any) => ({
        ...price,
        qmSearchCode,
        dateTime: new Date(price.date),
      }));

      await this.mongodb.batchUpsert(
        'qmPrices',
        processedPrices,
        ['qmSearchCode', 'date'] // Unique keys
      );

      // Find the latest price date
      const latestDate = processedPrices.reduce((latest: Date, price: any) => 
        price.dateTime > latest ? price.dateTime : latest, 
        new Date(0)
      );

      // Update symbol to track last price update
      const tracker = await getOperationTracker(this);
      await tracker.updateSymbolOperation(qmSearchCode, 'price_update', {
        status: 'success',
        lastRecordDate: latestDate,
        recordCount: priceData.length
      });

      this.logger.info('Prices updated successfully', { 
        qmSearchCode,
        priceCount: priceData.length,
        latestDate
      });

      return {
        success: true,
        qmSearchCode,
        message: `Prices updated for ${qmSearchCode}`,
        data: { 
          count: priceData.length,
          latestDate 
        }
      };
    } else {
      this.logger.warn('No price data returned from API', { qmSearchCode });
      return {
        success: false,
        qmSearchCode,
        message: `No price data found for qmSearchCode ${qmSearchCode}`
      };
    }

  } catch (error) {
    // Update session failure stats
    if (session.uuid) {
      await sessionManager.incrementFailedCalls(sessionId, session.uuid);
    }

    this.logger.error('Error fetching prices', { 
      qmSearchCode,
      error: error instanceof Error ? error.message : 'Unknown error'
    });
    
    // Track failure
    const tracker = await getOperationTracker(this);
    await tracker.updateSymbolOperation(qmSearchCode, 'price_update', {
      status: 'failure'
    });
    
    return {
      success: false,
      qmSearchCode,
      message: `Failed to fetch prices: ${error instanceof Error ? error.message : 'Unknown error'}`
    };
  }
}

/**
 * Schedule price updates for symbols that need refreshing
 */
export async function schedulePriceUpdates(
  this: QMHandler,
  input: {
    limit?: number;
    forceUpdate?: boolean;
  } = {},
  _context?: ExecutionContext
): Promise<{
  message: string;
  symbolsQueued: number;
  errors: number;
}> {
  const { limit = 1, forceUpdate = false } = input;
  const tracker = await getOperationTracker(this);
  
  this.logger.info('Scheduling price updates', { limit, forceUpdate });

  try {
    // Get symbols that need updating
    const staleSymbols = await tracker.getStaleSymbols('price_update', {
      minHoursSinceRun: forceUpdate ? 0 : 24, // Daily updates
      limit
    });

    if (staleSymbols.length === 0) {
      this.logger.info('No symbols need price updates');
      return {
        message: 'No symbols need price updates',
        symbolsQueued: 0,
        errors: 0
      };
    }

    this.logger.info(`Found ${staleSymbols.length} symbols needing price updates`);

    // Get full symbol data to include symbolId
    const symbolDocs = await this.mongodb.find('qmSymbols', {
      qmSearchCode: { $in: staleSymbols }
    }, {
      projection: { symbol: 1, symbolId: 1, qmSearchCode: 1 }
    });

    let queued = 0;
    let errors = 0;

    // Schedule individual update jobs for each symbol
    for (const doc of symbolDocs) {
      try {
        if (!doc.symbolId) {
          this.logger.warn(`Symbol ${doc.symbol} missing symbolId, skipping`);
          continue;
        }

        await this.scheduleOperation('update-prices', {
          qmSearchCode: doc.qmSearchCode
        }, {
          priority: 7, // High priority for price data
          delay: queued * 500 // 0.5 seconds between jobs
        });
        
        queued++;
      } catch (error) {
        this.logger.error(`Failed to schedule price update for ${doc.symbol}`, { error });
        errors++;
      }
    }

    this.logger.info('Price update scheduling completed', {
      symbolsQueued: queued,
      errors,
      total: staleSymbols.length
    });

    return {
      message: `Scheduled price updates for ${queued} symbols`,
      symbolsQueued: queued,
      errors
    };
  } catch (error) {
    this.logger.error('Price scheduling failed', { error });
    throw error;
  }
}