← back to Watches

utils/price-monitor.js

369 lines

/**
 * Automated Price Monitoring System
 * Daily scraping at 6 AM - stores historical data for chart analysis
 */

import PriceScraper from './price-scraper.js';
import fs from 'fs';
import path from 'path';
import { fileURLToPath } from 'url';
import pkg from 'pg';
const { Pool } = pkg;

const __filename = fileURLToPath(import.meta.url);
const __dirname = path.dirname(__filename);

class PriceMonitor {
  constructor() {
    this.scraper = new PriceScraper();
    this.pool = new Pool({
      host: process.env.DB_HOST || 'localhost',
      port: process.env.DB_PORT || 5432,
      database: process.env.DB_NAME || 'omega_watches',
      user: process.env.DB_USER || 'postgres',
      password: process.env.DB_PASSWORD || 'postgres'
    });

    this.csvDir = path.join(__dirname, '../data/price-history-csv');
    if (!fs.existsSync(this.csvDir)) {
      fs.mkdirSync(this.csvDir, { recursive: true });
    }
  }

  /**
   * Initialize database tables
   */
  async initDatabase() {
    try {
      // Create table first
      const createTableSQL = `
        CREATE TABLE IF NOT EXISTS price_history (
          id SERIAL PRIMARY KEY,
          watch_id VARCHAR(100) NOT NULL,
          watch_model VARCHAR(255) NOT NULL,
          watch_reference VARCHAR(100) NOT NULL,
          watch_series VARCHAR(100),

          -- Price data
          source VARCHAR(100) NOT NULL,
          price DECIMAL(10, 2),
          condition VARCHAR(50),

          -- Market statistics
          average_market_price DECIMAL(10, 2),
          lowest_price DECIMAL(10, 2),
          highest_price DECIMAL(10, 2),
          sample_size INTEGER,

          -- Metadata
          scrape_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
          scrape_date DATE DEFAULT CURRENT_DATE,

          -- Raw data
          raw_data JSONB,

          -- Indexes
          CONSTRAINT unique_daily_price UNIQUE (watch_id, scrape_date, source)
        );
      `;
      await this.pool.query(createTableSQL);

      // Create indexes after table exists - one at a time
      await this.pool.query('CREATE INDEX IF NOT EXISTS idx_watch_id ON price_history(watch_id)');
      await this.pool.query('CREATE INDEX IF NOT EXISTS idx_scrape_date ON price_history(scrape_date)');
      await this.pool.query('CREATE INDEX IF NOT EXISTS idx_watch_source ON price_history(watch_id, source)');
      await this.pool.query('CREATE INDEX IF NOT EXISTS idx_date_range ON price_history(watch_id, scrape_date)');

      // Create views after table and indexes exist
      const createViewsSQL = `
        -- Create aggregated daily view
        CREATE OR REPLACE VIEW daily_price_summary AS
        SELECT
          watch_id,
          watch_model,
          watch_reference,
          scrape_date,
          AVG(price) as avg_price,
          MIN(price) as min_price,
          MAX(price) as max_price,
          COUNT(*) as data_points,
          ARRAY_AGG(DISTINCT source) as sources
        FROM price_history
        WHERE price IS NOT NULL
        GROUP BY watch_id, watch_model, watch_reference, scrape_date
        ORDER BY scrape_date DESC;

        -- Create price trend view (30-day moving average)
        CREATE OR REPLACE VIEW price_trends AS
        SELECT
          watch_id,
          watch_model,
          scrape_date,
          avg_price,
          AVG(avg_price) OVER (
            PARTITION BY watch_id
            ORDER BY scrape_date
            ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
          ) as moving_avg_30d,
          LAG(avg_price, 1) OVER (PARTITION BY watch_id ORDER BY scrape_date) as prev_day_price,
          ((avg_price - LAG(avg_price, 1) OVER (PARTITION BY watch_id ORDER BY scrape_date))
           / LAG(avg_price, 1) OVER (PARTITION BY watch_id ORDER BY scrape_date) * 100) as daily_change_pct
        FROM daily_price_summary;
      `;
      await this.pool.query(createViewsSQL);

      console.log('✓ Database tables initialized');
    } catch (error) {
      console.error('✗ Database initialization failed:', error);
      throw error;
    }
  }

  /**
   * Store price data in PostgreSQL
   */
  async storePriceData(scrapedData) {
    const client = await this.pool.connect();

    try {
      await client.query('BEGIN');

      for (const pricePoint of scrapedData.prices) {
        if (!pricePoint.price) continue;

        const insertSQL = `
          INSERT INTO price_history (
            watch_id, watch_model, watch_reference, watch_series,
            source, price, condition,
            average_market_price, lowest_price, highest_price, sample_size,
            raw_data
          ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
          ON CONFLICT (watch_id, scrape_date, source)
          DO UPDATE SET
            price = EXCLUDED.price,
            condition = EXCLUDED.condition,
            average_market_price = EXCLUDED.average_market_price,
            raw_data = EXCLUDED.raw_data
        `;

        await client.query(insertSQL, [
          scrapedData.watch.id,
          scrapedData.watch.model,
          scrapedData.watch.reference,
          scrapedData.watch.series,
          pricePoint.source,
          pricePoint.price,
          pricePoint.condition || 'Unknown',
          scrapedData.averageMarketPrice || null,
          scrapedData.lowestPrice || null,
          scrapedData.highestPrice || null,
          scrapedData.prices.length,
          JSON.stringify(pricePoint)
        ]);
      }

      await client.query('COMMIT');
      console.log(`✓ Stored ${scrapedData.prices.length} price points in database`);

    } catch (error) {
      await client.query('ROLLBACK');
      console.error('✗ Database storage failed:', error);
      throw error;
    } finally {
      client.release();
    }
  }

  /**
   * Append to CSV file (stock-like format)
   */
  async appendToCSV(scrapedData) {
    const watchId = scrapedData.watch.id;
    const csvPath = path.join(this.csvDir, `${watchId}.csv`);

    const today = new Date().toISOString().split('T')[0];
    const avgPrice = scrapedData.averageMarketPrice || '';
    const low = scrapedData.lowestPrice || '';
    const high = scrapedData.highestPrice || '';
    const volume = scrapedData.prices.length;

    // Create CSV if doesn't exist
    if (!fs.existsSync(csvPath)) {
      const header = 'Date,Open,High,Low,Close,Volume,Model,Reference\n';
      fs.writeFileSync(csvPath, header);
    }

    // Read last price as "Open"
    const existingData = fs.readFileSync(csvPath, 'utf8');
    const lines = existingData.trim().split('\n');
    let lastClose = avgPrice;
    if (lines.length > 1) {
      const lastLine = lines[lines.length - 1].split(',');
      lastClose = lastLine[4] || avgPrice; // Use previous Close as today's Open
    }

    // Append new row (stock format: Date, Open, High, Low, Close, Volume)
    const row = `${today},${lastClose},${high},${low},${avgPrice},${volume},"${scrapedData.watch.model}","${scrapedData.watch.reference}"\n`;

    // Check if today's data already exists
    const todayExists = lines.some(line => line.startsWith(today));
    if (todayExists) {
      // Update today's row
      const updatedLines = lines.map(line => {
        if (line.startsWith(today)) {
          return row.trim();
        }
        return line;
      });
      fs.writeFileSync(csvPath, updatedLines.join('\n') + '\n');
      console.log(`✓ Updated CSV: ${csvPath}`);
    } else {
      // Append new row
      fs.appendFileSync(csvPath, row);
      console.log(`✓ Appended to CSV: ${csvPath}`);
    }
  }

  /**
   * Run daily price monitoring job
   */
  async runDailyMonitoring() {
    console.log('\n🕐 Starting daily price monitoring at', new Date().toLocaleString());
    console.log('=' .repeat(70));

    try {
      // Initialize database
      await this.initDatabase();

      // Load watches to monitor
      const dbPath = path.join(__dirname, '../data/watches.json');
      const database = JSON.parse(fs.readFileSync(dbPath, 'utf8'));

      const results = {
        timestamp: new Date().toISOString(),
        totalWatches: database.watches.length,
        successCount: 0,
        failureCount: 0,
        watches: []
      };

      // Monitor each watch
      for (const watch of database.watches) {
        try {
          console.log(`\n[${results.successCount + results.failureCount + 1}/${database.watches.length}] Processing: ${watch.model}`);

          // Scrape prices
          const scrapedData = await this.scraper.scrapeAllSources(watch);

          // Store in database
          await this.storePriceData(scrapedData);

          // Store in CSV
          await this.appendToCSV(scrapedData);

          results.successCount++;
          results.watches.push({
            id: watch.id,
            model: watch.model,
            status: 'success',
            pricePoints: scrapedData.prices.length,
            averagePrice: scrapedData.averageMarketPrice
          });

          // Rate limiting - wait 5 seconds between watches
          await new Promise(resolve => setTimeout(resolve, 5000));

        } catch (error) {
          console.error(`✗ Failed to monitor ${watch.model}:`, error.message);
          results.failureCount++;
          results.watches.push({
            id: watch.id,
            model: watch.model,
            status: 'failed',
            error: error.message
          });
        }
      }

      // Save daily report
      const reportPath = path.join(this.csvDir, `daily-report-${new Date().toISOString().split('T')[0]}.json`);
      fs.writeFileSync(reportPath, JSON.stringify(results, null, 2));

      console.log('\n' + '='.repeat(70));
      console.log('📊 Daily Monitoring Complete');
      console.log(`✓ Success: ${results.successCount}/${results.totalWatches}`);
      console.log(`✗ Failed: ${results.failureCount}/${results.totalWatches}`);
      console.log(`📄 Report saved: ${reportPath}`);

      return results;

    } catch (error) {
      console.error('✗ Daily monitoring failed:', error);
      throw error;
    } finally {
      await this.scraper.close();
      await this.pool.end();
    }
  }

  /**
   * Get price history for charting
   */
  async getPriceHistory(watchId, days = 90) {
    const query = `
      SELECT
        scrape_date as date,
        avg_price as close,
        min_price as low,
        max_price as high,
        data_points as volume,
        watch_model as model
      FROM daily_price_summary
      WHERE watch_id = $1
        AND scrape_date >= CURRENT_DATE - INTERVAL '${days} days'
      ORDER BY scrape_date ASC
    `;

    const result = await this.pool.query(query, [watchId]);
    return result.rows;
  }

  /**
   * Get price trends
   */
  async getPriceTrends(watchId) {
    const query = `
      SELECT
        watch_model,
        scrape_date,
        avg_price,
        moving_avg_30d,
        daily_change_pct
      FROM price_trends
      WHERE watch_id = $1
      ORDER BY scrape_date DESC
      LIMIT 90
    `;

    const result = await this.pool.query(query, [watchId]);
    return result.rows;
  }
}

export default PriceMonitor;

// CLI usage
if (import.meta.url === `file://${process.argv[1]}`) {
  const monitor = new PriceMonitor();

  (async () => {
    try {
      await monitor.runDailyMonitoring();
      process.exit(0);
    } catch (error) {
      console.error('Fatal error:', error);
      process.exit(1);
    }
  })();
}