384 lines
16 KiB
JavaScript
384 lines
16 KiB
JavaScript
const { outputProgress, formatElapsedTime, estimateRemaining, calculateRate, logError } = require('./utils/progress');
|
|
const { getConnection } = require('./utils/db');
|
|
|
|
async function calculateSalesForecasts(startTime, totalProducts, processedCount, isCancelled = false) {
|
|
const connection = await getConnection();
|
|
try {
|
|
if (isCancelled) {
|
|
outputProgress({
|
|
status: 'cancelled',
|
|
operation: 'Sales forecasts calculation cancelled',
|
|
current: processedCount,
|
|
total: totalProducts,
|
|
elapsed: formatElapsedTime(startTime),
|
|
remaining: null,
|
|
rate: calculateRate(startTime, processedCount),
|
|
percentage: ((processedCount / totalProducts) * 100).toFixed(1),
|
|
timing: {
|
|
start_time: new Date(startTime).toISOString(),
|
|
end_time: new Date().toISOString(),
|
|
elapsed_seconds: Math.round((Date.now() - startTime) / 1000)
|
|
}
|
|
});
|
|
return processedCount;
|
|
}
|
|
|
|
outputProgress({
|
|
status: 'running',
|
|
operation: 'Starting sales forecasts calculation',
|
|
current: processedCount,
|
|
total: totalProducts,
|
|
elapsed: formatElapsedTime(startTime),
|
|
remaining: estimateRemaining(startTime, processedCount, totalProducts),
|
|
rate: calculateRate(startTime, processedCount),
|
|
percentage: ((processedCount / totalProducts) * 100).toFixed(1),
|
|
timing: {
|
|
start_time: new Date(startTime).toISOString(),
|
|
end_time: new Date().toISOString(),
|
|
elapsed_seconds: Math.round((Date.now() - startTime) / 1000)
|
|
}
|
|
});
|
|
|
|
// First, create a temporary table for forecast dates
|
|
await connection.query(`
|
|
CREATE TEMPORARY TABLE IF NOT EXISTS temp_forecast_dates (
|
|
forecast_date DATE,
|
|
day_of_week INT,
|
|
month INT,
|
|
PRIMARY KEY (forecast_date)
|
|
)
|
|
`);
|
|
|
|
await connection.query(`
|
|
INSERT INTO temp_forecast_dates
|
|
SELECT
|
|
DATE_ADD(CURRENT_DATE, INTERVAL n DAY) as forecast_date,
|
|
DAYOFWEEK(DATE_ADD(CURRENT_DATE, INTERVAL n DAY)) as day_of_week,
|
|
MONTH(DATE_ADD(CURRENT_DATE, INTERVAL n DAY)) as month
|
|
FROM (
|
|
SELECT a.N + b.N * 10 as n
|
|
FROM
|
|
(SELECT 0 as N UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION
|
|
SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9) a,
|
|
(SELECT 0 as N UNION SELECT 1 UNION SELECT 2) b
|
|
ORDER BY n
|
|
LIMIT 31
|
|
) numbers
|
|
`);
|
|
|
|
processedCount = Math.floor(totalProducts * 0.92);
|
|
outputProgress({
|
|
status: 'running',
|
|
operation: 'Forecast dates prepared, calculating daily sales stats',
|
|
current: processedCount,
|
|
total: totalProducts,
|
|
elapsed: formatElapsedTime(startTime),
|
|
remaining: estimateRemaining(startTime, processedCount, totalProducts),
|
|
rate: calculateRate(startTime, processedCount),
|
|
percentage: ((processedCount / totalProducts) * 100).toFixed(1),
|
|
timing: {
|
|
start_time: new Date(startTime).toISOString(),
|
|
end_time: new Date().toISOString(),
|
|
elapsed_seconds: Math.round((Date.now() - startTime) / 1000)
|
|
}
|
|
});
|
|
|
|
if (isCancelled) return processedCount;
|
|
|
|
// Create temporary table for daily sales stats
|
|
await connection.query(`
|
|
CREATE TEMPORARY TABLE IF NOT EXISTS temp_daily_sales AS
|
|
SELECT
|
|
o.pid,
|
|
DAYOFWEEK(o.date) as day_of_week,
|
|
SUM(o.quantity) as daily_quantity,
|
|
SUM(o.price * o.quantity) as daily_revenue,
|
|
COUNT(DISTINCT DATE(o.date)) as day_count
|
|
FROM orders o
|
|
WHERE o.canceled = false
|
|
AND o.date >= DATE_SUB(CURRENT_DATE, INTERVAL 90 DAY)
|
|
GROUP BY o.pid, DAYOFWEEK(o.date)
|
|
`);
|
|
|
|
processedCount = Math.floor(totalProducts * 0.94);
|
|
outputProgress({
|
|
status: 'running',
|
|
operation: 'Daily sales stats calculated, preparing product stats',
|
|
current: processedCount,
|
|
total: totalProducts,
|
|
elapsed: formatElapsedTime(startTime),
|
|
remaining: estimateRemaining(startTime, processedCount, totalProducts),
|
|
rate: calculateRate(startTime, processedCount),
|
|
percentage: ((processedCount / totalProducts) * 100).toFixed(1),
|
|
timing: {
|
|
start_time: new Date(startTime).toISOString(),
|
|
end_time: new Date().toISOString(),
|
|
elapsed_seconds: Math.round((Date.now() - startTime) / 1000)
|
|
}
|
|
});
|
|
|
|
if (isCancelled) return processedCount;
|
|
|
|
// Create temporary table for product stats
|
|
await connection.query(`
|
|
CREATE TEMPORARY TABLE IF NOT EXISTS temp_product_stats AS
|
|
SELECT
|
|
pid,
|
|
AVG(daily_revenue) as overall_avg_revenue,
|
|
SUM(day_count) as total_days
|
|
FROM temp_daily_sales
|
|
GROUP BY pid
|
|
`);
|
|
|
|
processedCount = Math.floor(totalProducts * 0.96);
|
|
outputProgress({
|
|
status: 'running',
|
|
operation: 'Product stats prepared, calculating product-level forecasts',
|
|
current: processedCount,
|
|
total: totalProducts,
|
|
elapsed: formatElapsedTime(startTime),
|
|
remaining: estimateRemaining(startTime, processedCount, totalProducts),
|
|
rate: calculateRate(startTime, processedCount),
|
|
percentage: ((processedCount / totalProducts) * 100).toFixed(1),
|
|
timing: {
|
|
start_time: new Date(startTime).toISOString(),
|
|
end_time: new Date().toISOString(),
|
|
elapsed_seconds: Math.round((Date.now() - startTime) / 1000)
|
|
}
|
|
});
|
|
|
|
if (isCancelled) return processedCount;
|
|
|
|
// Calculate product-level forecasts
|
|
await connection.query(`
|
|
INSERT INTO sales_forecasts (
|
|
pid,
|
|
forecast_date,
|
|
forecast_units,
|
|
forecast_revenue,
|
|
confidence_level,
|
|
last_calculated_at
|
|
)
|
|
WITH daily_stats AS (
|
|
SELECT
|
|
ds.pid,
|
|
AVG(ds.daily_quantity) as avg_daily_qty,
|
|
STDDEV(ds.daily_quantity) as std_daily_qty,
|
|
COUNT(DISTINCT ds.day_count) as data_points,
|
|
SUM(ds.day_count) as total_days,
|
|
AVG(ds.daily_revenue) as avg_daily_revenue,
|
|
STDDEV(ds.daily_revenue) as std_daily_revenue,
|
|
MIN(ds.daily_quantity) as min_daily_qty,
|
|
MAX(ds.daily_quantity) as max_daily_qty,
|
|
AVG(ABS(ds.daily_quantity - LAG(ds.daily_quantity) OVER (PARTITION BY ds.pid ORDER BY ds.day_of_week))) as avg_daily_variance
|
|
FROM temp_daily_sales ds
|
|
GROUP BY ds.pid
|
|
HAVING AVG(ds.daily_quantity) > 0
|
|
)
|
|
SELECT
|
|
ds.pid,
|
|
fd.forecast_date,
|
|
GREATEST(0,
|
|
ROUND(
|
|
ds.avg_daily_qty *
|
|
(1 + COALESCE(sf.seasonality_factor, 0)) *
|
|
CASE
|
|
WHEN ds.std_daily_qty / NULLIF(ds.avg_daily_qty, 0) > 1.5 THEN 0.85
|
|
WHEN ds.std_daily_qty / NULLIF(ds.avg_daily_qty, 0) > 1.0 THEN 0.9
|
|
WHEN ds.std_daily_qty / NULLIF(ds.avg_daily_qty, 0) > 0.5 THEN 0.95
|
|
ELSE 1.0
|
|
END,
|
|
2
|
|
)
|
|
) as forecast_units,
|
|
GREATEST(0,
|
|
ROUND(
|
|
COALESCE(
|
|
CASE
|
|
WHEN ds.data_points >= 4 THEN ds.avg_daily_revenue
|
|
ELSE ps.overall_avg_revenue
|
|
END *
|
|
(1 + COALESCE(sf.seasonality_factor, 0)) *
|
|
CASE
|
|
WHEN ds.std_daily_revenue / NULLIF(ds.avg_daily_revenue, 0) > 1.5 THEN 0.85
|
|
WHEN ds.std_daily_revenue / NULLIF(ds.avg_daily_revenue, 0) > 1.0 THEN 0.9
|
|
WHEN ds.std_daily_revenue / NULLIF(ds.avg_daily_revenue, 0) > 0.5 THEN 0.95
|
|
ELSE 1.0
|
|
END,
|
|
0
|
|
),
|
|
2
|
|
)
|
|
) as forecast_revenue,
|
|
CASE
|
|
WHEN ds.total_days >= 60 AND ds.avg_daily_variance / NULLIF(ds.avg_daily_qty, 0) < 0.5 THEN 90
|
|
WHEN ds.total_days >= 60 THEN 85
|
|
WHEN ds.total_days >= 30 AND ds.avg_daily_variance / NULLIF(ds.avg_daily_qty, 0) < 0.5 THEN 80
|
|
WHEN ds.total_days >= 30 THEN 75
|
|
WHEN ds.total_days >= 14 AND ds.avg_daily_variance / NULLIF(ds.avg_daily_qty, 0) < 0.5 THEN 70
|
|
WHEN ds.total_days >= 14 THEN 65
|
|
ELSE 60
|
|
END as confidence_level,
|
|
NOW() as last_calculated_at
|
|
FROM daily_stats ds
|
|
JOIN temp_product_stats ps ON ds.pid = ps.pid
|
|
CROSS JOIN temp_forecast_dates fd
|
|
LEFT JOIN sales_seasonality sf ON fd.month = sf.month
|
|
GROUP BY ds.pid, fd.forecast_date, ps.overall_avg_revenue, sf.seasonality_factor
|
|
ON DUPLICATE KEY UPDATE
|
|
forecast_units = VALUES(forecast_units),
|
|
forecast_revenue = VALUES(forecast_revenue),
|
|
confidence_level = VALUES(confidence_level),
|
|
last_calculated_at = NOW()
|
|
`);
|
|
|
|
processedCount = Math.floor(totalProducts * 0.98);
|
|
outputProgress({
|
|
status: 'running',
|
|
operation: 'Product forecasts calculated, preparing category stats',
|
|
current: processedCount,
|
|
total: totalProducts,
|
|
elapsed: formatElapsedTime(startTime),
|
|
remaining: estimateRemaining(startTime, processedCount, totalProducts),
|
|
rate: calculateRate(startTime, processedCount),
|
|
percentage: ((processedCount / totalProducts) * 100).toFixed(1),
|
|
timing: {
|
|
start_time: new Date(startTime).toISOString(),
|
|
end_time: new Date().toISOString(),
|
|
elapsed_seconds: Math.round((Date.now() - startTime) / 1000)
|
|
}
|
|
});
|
|
|
|
if (isCancelled) return processedCount;
|
|
|
|
// Create temporary table for category stats
|
|
await connection.query(`
|
|
CREATE TEMPORARY TABLE IF NOT EXISTS temp_category_sales AS
|
|
SELECT
|
|
pc.cat_id,
|
|
DAYOFWEEK(o.date) as day_of_week,
|
|
SUM(o.quantity) as daily_quantity,
|
|
SUM(o.price * o.quantity) as daily_revenue,
|
|
COUNT(DISTINCT DATE(o.date)) as day_count
|
|
FROM orders o
|
|
JOIN product_categories pc ON o.pid = pc.pid
|
|
WHERE o.canceled = false
|
|
AND o.date >= DATE_SUB(CURRENT_DATE, INTERVAL 90 DAY)
|
|
GROUP BY pc.cat_id, DAYOFWEEK(o.date)
|
|
`);
|
|
|
|
await connection.query(`
|
|
CREATE TEMPORARY TABLE IF NOT EXISTS temp_category_stats AS
|
|
SELECT
|
|
cat_id,
|
|
AVG(daily_revenue) as overall_avg_revenue,
|
|
SUM(day_count) as total_days
|
|
FROM temp_category_sales
|
|
GROUP BY cat_id
|
|
`);
|
|
|
|
processedCount = Math.floor(totalProducts * 0.99);
|
|
outputProgress({
|
|
status: 'running',
|
|
operation: 'Category stats prepared, calculating category-level forecasts',
|
|
current: processedCount,
|
|
total: totalProducts,
|
|
elapsed: formatElapsedTime(startTime),
|
|
remaining: estimateRemaining(startTime, processedCount, totalProducts),
|
|
rate: calculateRate(startTime, processedCount),
|
|
percentage: ((processedCount / totalProducts) * 100).toFixed(1),
|
|
timing: {
|
|
start_time: new Date(startTime).toISOString(),
|
|
end_time: new Date().toISOString(),
|
|
elapsed_seconds: Math.round((Date.now() - startTime) / 1000)
|
|
}
|
|
});
|
|
|
|
if (isCancelled) return processedCount;
|
|
|
|
// Calculate category-level forecasts
|
|
await connection.query(`
|
|
INSERT INTO category_forecasts (
|
|
category_id,
|
|
forecast_date,
|
|
forecast_units,
|
|
forecast_revenue,
|
|
confidence_level,
|
|
last_calculated_at
|
|
)
|
|
SELECT
|
|
cs.cat_id as category_id,
|
|
fd.forecast_date,
|
|
GREATEST(0,
|
|
AVG(cs.daily_quantity) *
|
|
(1 + COALESCE(sf.seasonality_factor, 0))
|
|
) as forecast_units,
|
|
GREATEST(0,
|
|
COALESCE(
|
|
CASE
|
|
WHEN SUM(cs.day_count) >= 4 THEN AVG(cs.daily_revenue)
|
|
ELSE ct.overall_avg_revenue
|
|
END *
|
|
(1 + COALESCE(sf.seasonality_factor, 0)) *
|
|
(0.95 + (RAND() * 0.1)),
|
|
0
|
|
)
|
|
) as forecast_revenue,
|
|
CASE
|
|
WHEN ct.total_days >= 60 THEN 90
|
|
WHEN ct.total_days >= 30 THEN 80
|
|
WHEN ct.total_days >= 14 THEN 70
|
|
ELSE 60
|
|
END as confidence_level,
|
|
NOW() as last_calculated_at
|
|
FROM temp_category_sales cs
|
|
JOIN temp_category_stats ct ON cs.cat_id = ct.cat_id
|
|
CROSS JOIN temp_forecast_dates fd
|
|
LEFT JOIN sales_seasonality sf ON fd.month = sf.month
|
|
GROUP BY cs.cat_id, fd.forecast_date, ct.overall_avg_revenue, ct.total_days, sf.seasonality_factor
|
|
HAVING AVG(cs.daily_quantity) > 0
|
|
ON DUPLICATE KEY UPDATE
|
|
forecast_units = VALUES(forecast_units),
|
|
forecast_revenue = VALUES(forecast_revenue),
|
|
confidence_level = VALUES(confidence_level),
|
|
last_calculated_at = NOW()
|
|
`);
|
|
|
|
// Clean up temporary tables
|
|
await connection.query(`
|
|
DROP TEMPORARY TABLE IF EXISTS temp_forecast_dates;
|
|
DROP TEMPORARY TABLE IF EXISTS temp_daily_sales;
|
|
DROP TEMPORARY TABLE IF EXISTS temp_product_stats;
|
|
DROP TEMPORARY TABLE IF EXISTS temp_category_sales;
|
|
DROP TEMPORARY TABLE IF EXISTS temp_category_stats;
|
|
`);
|
|
|
|
processedCount = Math.floor(totalProducts * 1.0);
|
|
outputProgress({
|
|
status: 'running',
|
|
operation: 'Category forecasts calculated and temporary tables cleaned up',
|
|
current: processedCount,
|
|
total: totalProducts,
|
|
elapsed: formatElapsedTime(startTime),
|
|
remaining: estimateRemaining(startTime, processedCount, totalProducts),
|
|
rate: calculateRate(startTime, processedCount),
|
|
percentage: ((processedCount / totalProducts) * 100).toFixed(1),
|
|
timing: {
|
|
start_time: new Date(startTime).toISOString(),
|
|
end_time: new Date().toISOString(),
|
|
elapsed_seconds: Math.round((Date.now() - startTime) / 1000)
|
|
}
|
|
});
|
|
|
|
return processedCount;
|
|
} catch (error) {
|
|
logError(error, 'Error calculating sales forecasts');
|
|
throw error;
|
|
} finally {
|
|
if (connection) {
|
|
connection.release();
|
|
}
|
|
}
|
|
}
|
|
|
|
module.exports = calculateSalesForecasts;
|