qm scaffolding done

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

@@ -0,0 +1,249 @@
+/**
+ * QM Prices Actions - Fetch and update daily price data
+ */
+
+import type { BaseHandler, ExecutionContext } from '@stock-bot/handlers';
+import { QM_CONFIG, QM_SESSION_IDS } from '../shared/config';
+import { QMSessionManager } from '../shared/session-manager';
+import { QMOperationTracker } from '../shared/operation-tracker';
+
+// Cache tracker instance
+let operationTracker: QMOperationTracker | null = null;
+
+/**
+ * Get or initialize the operation tracker
+ */
+async function getOperationTracker(handler: BaseHandler): 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: BaseHandler,
+  input: {
+    symbol: string;
+    symbolId: number;
+  },
+  _context?: ExecutionContext
+): Promise<{
+  success: boolean;
+  symbol: string;
+  message: string;
+  data?: any;
+}> {
+  const { symbol, symbolId } = input;
+  
+  this.logger.info('Fetching daily prices', { symbol, symbolId });
+
+  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.LOOKUP; // TODO: Update with correct session ID
+  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({
+      symbol: symbol,
+      symbolId: symbolId.toString(),
+      qmodTool: 'DailyPrices',
+      webmasterId: '500',
+      days: '30' // Get last 30 days
+    });
+    
+    // TODO: Update with correct prices endpoint
+    const apiUrl = `${QM_CONFIG.BASE_URL}/datatool/prices.json?${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 priceData = await response.json();
+    
+    // Update session success stats
+    await sessionManager.incrementSuccessfulCalls(sessionId, session.uuid);
+
+    // Process and store price data
+    if (priceData && priceData.length > 0) {
+      // Store prices in a separate collection
+      const processedPrices = priceData.map((price: any) => ({
+        ...price,
+        symbol,
+        symbolId,
+        date: new Date(price.date),
+        updated_at: new Date()
+      }));
+
+      await this.mongodb.batchUpsert(
+        'qmPrices',
+        processedPrices,
+        ['symbol', 'date'] // Unique keys
+      );
+
+      // Find the latest price date
+      const latestDate = processedPrices.reduce((latest: Date, price: any) => 
+        price.date > latest ? price.date : latest, 
+        new Date(0)
+      );
+
+      // Update symbol to track last price update
+      const tracker = await getOperationTracker(this);
+      await tracker.updateSymbolOperation(symbol, 'price_update', {
+        status: 'success',
+        lastRecordDate: latestDate,
+        recordCount: priceData.length
+      });
+
+      this.logger.info('Prices updated successfully', { 
+        symbol,
+        priceCount: priceData.length,
+        latestDate
+      });
+
+      return {
+        success: true,
+        symbol,
+        message: `Prices updated for ${symbol}`,
+        data: { 
+          count: priceData.length,
+          latestDate 
+        }
+      };
+    } else {
+      this.logger.warn('No price data returned from API', { symbol });
+      return {
+        success: false,
+        symbol,
+        message: `No price data found for symbol ${symbol}`
+      };
+    }
+
+  } catch (error) {
+    // Update session failure stats
+    if (session.uuid) {
+      await sessionManager.incrementFailedCalls(sessionId, session.uuid);
+    }
+
+    this.logger.error('Error fetching prices', { 
+      symbol,
+      error: error instanceof Error ? error.message : 'Unknown error'
+    });
+    
+    // Track failure
+    const tracker = await getOperationTracker(this);
+    await tracker.updateSymbolOperation(symbol, 'price_update', {
+      status: 'failure'
+    });
+    
+    return {
+      success: false,
+      symbol,
+      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: BaseHandler,
+  input: {
+    limit?: number;
+    forceUpdate?: boolean;
+  } = {},
+  _context?: ExecutionContext
+): Promise<{
+  message: string;
+  symbolsQueued: number;
+  errors: number;
+}> {
+  const { limit = 100, 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', {
+      symbol: { $in: staleSymbols }
+    }, {
+      projection: { symbol: 1, symbolId: 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', {
+          symbol: doc.symbol,
+          symbolId: doc.symbolId
+        }, {
+          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;
+  }
+}