`services/price-aggregator.js`

272 lines
/**
 * Price Aggregation Service
 * Combines prices from multiple sources with weighting and confidence scoring
 */

import waybackScraper from './wayback-scraper.js';
import githubDataset from './github-dataset.js';
import chrono24Scraper from './chrono24-scraper.js';
import pool from '../database/connection.js';

// Source weights (higher = more trusted)
const SOURCE_WEIGHTS = {
  official: 1.0,      // Official Omega prices
  chrono24: 0.9,      // Live marketplace
  github: 0.8,        // Curated datasets
  wayback: 0.7,       // Historical archives
  user: 0.5           // User-submitted
};

// Recency weights (exponential decay)
const RECENCY_HALF_LIFE_DAYS = 30;

/**
 * Calculate recency weight
 * @param {string} dateStr - Date string
 * @returns {number} Weight between 0 and 1
 */
function recencyWeight(dateStr) {
  const daysSince = (Date.now() - new Date(dateStr).getTime()) / (24 * 60 * 60 * 1000);
  return Math.pow(0.5, daysSince / RECENCY_HALF_LIFE_DAYS);
}

/**
 * Calculate weighted statistics
 * @param {array} prices - Array of { price, weight }
 * @returns {object} Statistics
 */
function weightedStatistics(prices) {
  if (prices.length === 0) {
    return { avg: 0, median: 0, min: 0, max: 0, stdDev: 0, confidence: 0 };
  }

  const totalWeight = prices.reduce((sum, p) => sum + p.weight, 0);
  const weightedSum = prices.reduce((sum, p) => sum + p.price * p.weight, 0);
  const weightedAvg = weightedSum / totalWeight;

  // Weighted standard deviation
  const weightedVariance = prices.reduce((sum, p) => {
    return sum + p.weight * Math.pow(p.price - weightedAvg, 2);
  }, 0) / totalWeight;
  const stdDev = Math.sqrt(weightedVariance);

  // Sort for median and range
  const sorted = [...prices].sort((a, b) => a.price - b.price);

  // Confidence based on data count and weight consistency
  const confidence = Math.min(
    5,
    Math.round(1 + Math.log2(prices.length + 1) + (totalWeight / prices.length))
  );

  return {
    avg: Math.round(weightedAvg),
    median: sorted[Math.floor(sorted.length / 2)].price,
    min: sorted[0].price,
    max: sorted[sorted.length - 1].price,
    stdDev: Math.round(stdDev),
    confidence,
    dataPoints: prices.length,
    totalWeight: Math.round(totalWeight * 100) / 100
  };
}

/**
 * Fetch prices from all sources for a watch reference
 * @param {string} reference - Watch reference number
 * @returns {array} Array of price records with weights
 */
export async function fetchAllPrices(reference) {
  const allPrices = [];

  // Parallel fetch from all sources
  const [waybackPrices, githubPrices, chrono24Prices] = await Promise.all([
    waybackScraper.scrapeHistoricalPrices(reference).catch(() => []),
    githubDataset.fetchOmegaPrices().then(prices =>
      prices.filter(p => p.reference?.includes(reference))
    ).catch(() => []),
    chrono24Scraper.scrapeOmegaListings(reference).catch(() => [])
  ]);

  // Process Wayback prices
  for (const p of waybackPrices) {
    allPrices.push({
      price: p.price,
      source: 'wayback',
      date: p.date,
      weight: SOURCE_WEIGHTS.wayback * recencyWeight(p.date),
      raw: p
    });
  }

  // Process GitHub prices
  for (const p of githubPrices) {
    allPrices.push({
      price: p.price,
      source: 'github',
      date: p.date || new Date().toISOString().slice(0, 10),
      weight: SOURCE_WEIGHTS.github * recencyWeight(p.date || new Date().toISOString()),
      raw: p
    });
  }

  // Process Chrono24 prices
  for (const p of chrono24Prices) {
    allPrices.push({
      price: p.price,
      source: 'chrono24',
      date: p.scrapedAt?.slice(0, 10) || new Date().toISOString().slice(0, 10),
      weight: SOURCE_WEIGHTS.chrono24 * recencyWeight(p.scrapedAt || new Date().toISOString()),
      condition: p.condition,
      raw: p
    });
  }

  return allPrices;
}

/**
 * Aggregate prices for a watch reference
 * @param {string} reference - Watch reference number
 * @returns {object} Aggregated price data
 */
export async function aggregatePrices(reference) {
  const allPrices = await fetchAllPrices(reference);

  if (allPrices.length === 0) {
    return {
      reference,
      hasData: false,
      message: 'No price data available'
    };
  }

  const stats = weightedStatistics(allPrices);

  // Group by source for breakdown
  const bySource = {};
  for (const p of allPrices) {
    if (!bySource[p.source]) {
      bySource[p.source] = [];
    }
    bySource[p.source].push(p.price);
  }

  const sourceBreakdown = {};
  for (const [source, prices] of Object.entries(bySource)) {
    sourceBreakdown[source] = {
      count: prices.length,
      avg: Math.round(prices.reduce((a, b) => a + b, 0) / prices.length),
      min: Math.min(...prices),
      max: Math.max(...prices)
    };
  }

  return {
    reference,
    hasData: true,
    currentMarketPrice: stats.avg,
    priceRange: { min: stats.min, max: stats.max },
    median: stats.median,
    stdDev: stats.stdDev,
    confidence: stats.confidence,
    confidenceLabel: getConfidenceLabel(stats.confidence),
    dataPoints: stats.dataPoints,
    sourceBreakdown,
    lastUpdated: new Date().toISOString()
  };
}

/**
 * Get confidence label
 */
function getConfidenceLabel(confidence) {
  if (confidence >= 4) return 'high';
  if (confidence >= 2) return 'medium';
  return 'low';
}

/**
 * Store aggregated prices in database
 * @param {string} watchId - Watch UUID
 * @param {object} aggregation - Aggregated price data
 */
export async function storeAggregation(watchId, aggregation) {
  try {
    await pool.query(
      `INSERT INTO aggregated_prices (watch_id, avg_price, min_price, max_price, median_price,
                                       confidence, data_points, sources, updated_at)
       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW())
       ON CONFLICT (watch_id) DO UPDATE SET
         avg_price = $2, min_price = $3, max_price = $4, median_price = $5,
         confidence = $6, data_points = $7, sources = $8, updated_at = NOW()`,
      [
        watchId,
        aggregation.currentMarketPrice,
        aggregation.priceRange?.min,
        aggregation.priceRange?.max,
        aggregation.median,
        aggregation.confidence,
        aggregation.dataPoints,
        JSON.stringify(aggregation.sourceBreakdown)
      ]
    );
    return true;
  } catch (error) {
    console.error('Failed to store aggregation:', error.message);
    return false;
  }
}

/**
 * Get stored aggregation from database
 * @param {string} watchId - Watch UUID
 * @returns {object} Stored aggregation or null
 */
export async function getStoredAggregation(watchId) {
  try {
    const result = await pool.query(
      `SELECT * FROM aggregated_prices WHERE watch_id = $1`,
      [watchId]
    );
    return result.rows[0] || null;
  } catch (error) {
    console.error('Failed to get aggregation:', error.message);
    return null;
  }
}

/**
 * Aggregate prices for all watches
 * @param {array} watches - Array of watch records
 * @returns {array} Array of aggregation results
 */
export async function aggregateAllWatches(watches) {
  const results = [];

  for (const watch of watches) {
    try {
      console.log(`Aggregating prices for ${watch.reference}...`);
      const aggregation = await aggregatePrices(watch.reference);
      results.push({ watchId: watch.id, ...aggregation });

      // Rate limit between watches
      await new Promise(resolve => setTimeout(resolve, 1000));
    } catch (error) {
      console.error(`Failed to aggregate ${watch.reference}:`, error.message);
      results.push({ watchId: watch.id, error: error.message });
    }
  }

  return results;
}

export default {
  fetchAllPrices,
  aggregatePrices,
  storeAggregation,
  getStoredAggregation,
  aggregateAllWatches,
  SOURCE_WEIGHTS
};