120 lines
3.5 KiB
Python
120 lines
3.5 KiB
Python
"""销量预测/分级/新品判断函数"""
|
||
from datetime import datetime
|
||
from dateutil.relativedelta import relativedelta
|
||
from services.db import query
|
||
from services.cache import RefreshableCache, register_cache
|
||
|
||
|
||
def is_new_product(created_at_str):
|
||
"""判断单品是否为新品（创建时间在5个月以内）"""
|
||
if not created_at_str:
|
||
return '返单品'
|
||
try:
|
||
if isinstance(created_at_str, str):
|
||
created = datetime.strptime(created_at_str[:10], '%Y-%m-%d')
|
||
else:
|
||
created = created_at_str
|
||
if hasattr(created, 'strftime'):
|
||
five_months_ago = datetime.now() - relativedelta(months=5)
|
||
return '新品' if created >= five_months_ago else '返单品'
|
||
return '返单品'
|
||
except:
|
||
return '返单品'
|
||
|
||
|
||
def _load_global_config():
|
||
"""加载全局配置"""
|
||
try:
|
||
gc_rows = query("SELECT config_key, config_value FROM global_config")
|
||
return {r['config_key']: r['config_value'] for r in gc_rows}
|
||
except:
|
||
return {
|
||
'safety_stock_days': '7',
|
||
'suggested_return_days': '15',
|
||
}
|
||
|
||
|
||
# 全局配置缓存（文件共享，所有 worker 共用）
|
||
_global_config_cache_obj = RefreshableCache(maxsize=1)
|
||
register_cache(_global_config_cache_obj)
|
||
|
||
|
||
@_global_config_cache_obj.wrap
|
||
def get_global_config():
|
||
"""缓存版全局配置（支持主动刷新）"""
|
||
return _load_global_config()
|
||
|
||
|
||
def refresh_global_config():
|
||
"""刷新全局配置缓存"""
|
||
_global_config_cache_obj.refresh()
|
||
|
||
|
||
def _sales_predict_config():
|
||
"""获取日销量预测基础配置参数"""
|
||
rows = query("SELECT * FROM sales_predict_config ORDER BY id")
|
||
return rows
|
||
|
||
|
||
# 预测配置缓存
|
||
predict_config_cache = RefreshableCache(maxsize=1)
|
||
register_cache(predict_config_cache)
|
||
|
||
|
||
@predict_config_cache.wrap
|
||
def sales_predict_config_cached():
|
||
"""缓存版预测配置（支持主动刷新）"""
|
||
return _sales_predict_config()
|
||
|
||
|
||
def refresh_sales_predict_config():
|
||
"""刷新销量预测配置缓存，使保存后的权重立即对所有 worker 生效。"""
|
||
predict_config_cache.refresh()
|
||
|
||
|
||
def get_product_grade(monthly_sales, product_name='', product_type=''):
|
||
"""根据月销量进行产品分级"""
|
||
# 赠品/晒图卡不参与评级
|
||
if product_type == '赠品' or '晒图卡' in (product_name or ''):
|
||
return '不参与评级'
|
||
|
||
# 小样最高C级
|
||
if product_type == '小样':
|
||
return 'C级'
|
||
|
||
try:
|
||
ms = float(monthly_sales)
|
||
except:
|
||
return 'C级'
|
||
if ms >= 10000:
|
||
return 'S级'
|
||
elif ms >= 5000:
|
||
return 'A级'
|
||
elif ms >= 3000:
|
||
return 'B级'
|
||
else:
|
||
return 'C级'
|
||
|
||
|
||
def compute_daily_forecast(row, sales_predict_map):
|
||
"""计算日销量预测值"""
|
||
avg3 = float(row.get('3天平均销量', 0) or 0)
|
||
avg7 = float(row.get('7天平均销量', 0) or 0)
|
||
avg14 = float(row.get('14天平均销量', 0) or 0)
|
||
max30 = float(row.get('30天最高销量', 0) or 0)
|
||
trend = row.get('销售趋势', '平稳趋势')
|
||
|
||
if avg3 == 0 and avg7 == 0 and avg14 == 0 and max30 == 0:
|
||
return 0
|
||
|
||
# 根据趋势类型获取预测系数
|
||
cfg = sales_predict_map.get(trend, {})
|
||
w3 = float(cfg.get('avg_3d', 0.5))
|
||
w7 = float(cfg.get('avg_7d', 0.3))
|
||
w14 = float(cfg.get('avg_14d', 0.2))
|
||
adj = float(cfg.get('adjust_factor', 1.0))
|
||
|
||
# 计算加权预测值
|
||
forecast = (avg3 * w3 + avg7 * w7 + avg14 * w14) * adj
|
||
return round(forecast)
|