341 lines
16 KiB
Python
341 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""AI分析P2：情景模拟、未来风险预测和只读行动草稿。
|
||
|
||
所有结果均为模拟/草稿，不写入订单、库存或待下单业务表。
|
||
"""
|
||
from collections import defaultdict
|
||
from datetime import date, timedelta
|
||
import math
|
||
|
||
from services.ai_datasources import AI_DATASOURCES, datasource_meta
|
||
from services.ai_insights import get_operational_insights, filter_insights, position_profile
|
||
from services.simple_cache import get as cache_get, set as cache_set
|
||
|
||
FORECAST_CACHE_PREFIX = 'ai_p2_forecast_v1'
|
||
|
||
|
||
def _num(value, default=0.0):
|
||
try:
|
||
return float(value or 0)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def _int(value):
|
||
return int(round(_num(value)))
|
||
|
||
|
||
def _first(key, params):
|
||
result = AI_DATASOURCES[key]['handler'](params) or {}
|
||
rows = result.get('rows') or []
|
||
return rows[0] if rows else {}
|
||
|
||
|
||
def _rows(key, params=None):
|
||
result = AI_DATASOURCES[key]['handler'](params or {'_no_limit': True}) or {}
|
||
return result.get('rows') or []
|
||
|
||
|
||
def _risk_level(support_days, threshold_days, daily_sales):
|
||
if daily_sales <= 0:
|
||
return '低'
|
||
if support_days < max(threshold_days, 1):
|
||
return '高'
|
||
if support_days < threshold_days + 15:
|
||
return '中'
|
||
return '低'
|
||
|
||
|
||
def _round_qty(value):
|
||
return max(0, int(round(_num(value))))
|
||
|
||
|
||
def simulate_product(product_code, sales_change_pct=0, supplier_delay_days=0,
|
||
planned_order_qty=0, horizon_days=30, allowed_sources=None):
|
||
"""对单个SKU进行只读情景模拟。"""
|
||
product_code = str(product_code or '').strip()
|
||
if not product_code:
|
||
raise ValueError('请输入产品编码')
|
||
sales_change_pct = max(-100.0, min(_num(sales_change_pct), 500.0))
|
||
supplier_delay_days = max(0, min(_int(supplier_delay_days), 180))
|
||
planned_order_qty = max(0, min(_int(planned_order_qty), 100000000))
|
||
horizon_days = max(1, min(_int(horizon_days) or 30, 365))
|
||
params = {'product_code': product_code, 'limit': 200}
|
||
allowed = set(allowed_sources or AI_DATASOURCES.keys())
|
||
if not {'product_profile', 'planning_metrics'}.issubset(allowed):
|
||
raise PermissionError('需要产品信息和计划预测查看权限才能进行模拟')
|
||
|
||
profile = _first('product_profile', params)
|
||
planning = _first('planning_metrics', params)
|
||
sales = _first('product_sales', params) if 'product_sales' in allowed else {}
|
||
inventory = _first('product_inventory', params) if 'product_inventory' in allowed else {}
|
||
orders = _first('orders_in_transit', params) if 'orders_in_transit' in allowed else {}
|
||
bom_rows = _rows('bom_supply_chain', params) if 'bom_supply_chain' in allowed else []
|
||
if not profile and not planning:
|
||
raise ValueError('未找到该产品，请核对产品编码')
|
||
|
||
code = str(planning.get('product_code') or profile.get('product_code') or product_code)
|
||
product_type = str(profile.get('产品类型') or '')
|
||
lead = _int(profile.get('生产货期'))
|
||
logistics = _int(profile.get('物流时间'))
|
||
safety = 0 if product_type == '现货' else 7
|
||
base_threshold = lead + logistics + safety
|
||
scenario_threshold = base_threshold + supplier_delay_days
|
||
base_daily = _num(planning.get('日销量预测'))
|
||
if base_daily <= 0:
|
||
base_daily = _num(sales.get('30天平均销量'))
|
||
scenario_daily = max(0.0, base_daily * (1 + sales_change_pct / 100.0))
|
||
|
||
quantities = {
|
||
'库存': _int(planning.get('库存', inventory.get('库存'))),
|
||
'加工厂库存': _int(planning.get('加工厂库存', inventory.get('加工厂库存'))),
|
||
'生产在途': _int(planning.get('生产在途', inventory.get('生产在途'))),
|
||
'采购在途': _int(planning.get('采购在途', inventory.get('采购在途'))),
|
||
}
|
||
total_supply = sum(quantities.values())
|
||
base_support = round(total_supply / base_daily, 1) if base_daily > 0 else 999
|
||
scenario_support = round(total_supply / scenario_daily, 1) if scenario_daily > 0 else 999
|
||
support_after_order = round((total_supply + planned_order_qty) / scenario_daily, 1) if scenario_daily > 0 else 999
|
||
|
||
if scenario_daily <= 0 or scenario_support >= scenario_threshold:
|
||
recommended_order = 0
|
||
else:
|
||
recommended_order = _round_qty(scenario_daily * scenario_threshold)
|
||
shortage_date = None
|
||
latest_order_date = None
|
||
if scenario_daily > 0 and scenario_support < 999:
|
||
shortage_day = date.today() + timedelta(days=max(0, int(math.floor(scenario_support))))
|
||
shortage_date = shortage_day.isoformat()
|
||
latest_order_date = (shortage_day - timedelta(days=scenario_threshold)).isoformat()
|
||
|
||
target_qty = planned_order_qty or recommended_order
|
||
bom_gaps = []
|
||
for material in bom_rows:
|
||
unit_qty = _num(material.get('单位用量'))
|
||
supported = _int(material.get('可支撑成品数量'))
|
||
if target_qty <= 0 or unit_qty <= 0 or supported >= target_qty:
|
||
continue
|
||
finished_gap = target_qty - supported
|
||
bom_gaps.append({
|
||
'material_code': material.get('material_code') or '',
|
||
'material_name': material.get('物料名称') or '',
|
||
'supplier': material.get('物料供应商') or '',
|
||
'unit_qty': unit_qty,
|
||
'support_finished_qty': supported,
|
||
'target_finished_qty': target_qty,
|
||
'finished_gap': finished_gap,
|
||
'material_gap': int(math.ceil(finished_gap * unit_qty)),
|
||
})
|
||
bom_gaps.sort(key=lambda x: x['finished_gap'], reverse=True)
|
||
|
||
base_level = _risk_level(base_support, base_threshold, base_daily)
|
||
scenario_level = _risk_level(scenario_support, scenario_threshold, scenario_daily)
|
||
if scenario_daily <= 0:
|
||
conclusion = '模拟后日销量为0，按当前规则无需新增下单。'
|
||
elif scenario_support >= scenario_threshold:
|
||
conclusion = f'模拟后总支撑天数{scenario_support:.1f}天，不低于补货阈值{scenario_threshold}天，暂不建议新增下单。'
|
||
else:
|
||
conclusion = f'模拟后总支撑天数{scenario_support:.1f}天，低于补货阈值{scenario_threshold}天，建议评估下单{recommended_order}件。'
|
||
if planned_order_qty:
|
||
conclusion += f' 若下单{planned_order_qty}件，下单后预计可支撑{support_after_order:.1f}天。'
|
||
if bom_gaps:
|
||
conclusion += f' 但有{len(bom_gaps)}项BOM物料不足，需先处理物料缺口。'
|
||
|
||
return {
|
||
'product_code': code,
|
||
'product_name': profile.get('货品名称') or '',
|
||
'brand': profile.get('品牌') or planning.get('品牌') or '',
|
||
'product_type': product_type,
|
||
'inputs': {
|
||
'sales_change_pct': sales_change_pct,
|
||
'supplier_delay_days': supplier_delay_days,
|
||
'planned_order_qty': planned_order_qty,
|
||
'horizon_days': horizon_days,
|
||
},
|
||
'base': {
|
||
'daily_sales': round(base_daily, 2), 'support_days': base_support,
|
||
'threshold_days': base_threshold, 'risk_level': base_level,
|
||
'suggested_order_qty': _int(planning.get('建议下单数量')),
|
||
},
|
||
'scenario': {
|
||
'daily_sales': round(scenario_daily, 2), 'support_days': scenario_support,
|
||
'threshold_days': scenario_threshold, 'risk_level': scenario_level,
|
||
'recommended_order_qty': recommended_order,
|
||
'support_after_order_days': support_after_order,
|
||
'estimated_shortage_date': shortage_date,
|
||
'latest_order_date': latest_order_date,
|
||
'horizon_demand': _round_qty(scenario_daily * horizon_days),
|
||
},
|
||
'inventory': quantities | {
|
||
'合计供应数量': total_supply,
|
||
'国内未到数量': _int(orders.get('国内未到数量')),
|
||
'海外未到数量': _int(orders.get('海外未到数量')),
|
||
'包材未到数量': _int(orders.get('包材未到数量')),
|
||
},
|
||
'bom_gaps': bom_gaps[:30],
|
||
'conclusion': conclusion,
|
||
'assumptions': [
|
||
'模拟不会修改库存、订单、计划预测或待下单数据。',
|
||
'供应商延期天数计入补货阈值，用于压力测试。',
|
||
'销量变化按当前日销量预测等比例调整。',
|
||
'BOM缺口按目标成品数量与当前可支撑成品数量比较。',
|
||
],
|
||
'data_dates': {
|
||
'inventory': (datasource_meta('product_inventory').get('data_date') or '') if 'product_inventory' in allowed else '',
|
||
'sales': (datasource_meta('product_sales').get('data_date') or '') if 'product_sales' in allowed else '',
|
||
},
|
||
}
|
||
|
||
|
||
def forecast_supply_risks(horizon_days=30, limit=100, allowed_sources=None):
|
||
"""预测未来指定天数内的缺货/补货风险。"""
|
||
horizon_days = max(1, min(_int(horizon_days) or 30, 365))
|
||
limit = max(1, min(_int(limit) or 100, 300))
|
||
allowed = set(allowed_sources or AI_DATASOURCES.keys())
|
||
if not {'planning_metrics', 'product_profile'}.issubset(allowed):
|
||
raise PermissionError('需要计划预测和产品信息查看权限才能进行风险预测')
|
||
include_bom = 'bom_supply_chain' in allowed
|
||
cache_key = f'{FORECAST_CACHE_PREFIX}:{horizon_days}:{limit}:{int(include_bom)}'
|
||
cached = cache_get(cache_key)
|
||
if cached is not None:
|
||
return cached
|
||
|
||
planning = _rows('planning_metrics')
|
||
profiles = {str(r.get('product_code') or ''): r for r in _rows('product_profile')}
|
||
bom = defaultdict(list)
|
||
if include_bom:
|
||
for row in _rows('bom_supply_chain'):
|
||
bom[str(row.get('product_code') or '')].append(row)
|
||
rows = []
|
||
for row in planning:
|
||
code = str(row.get('product_code') or '')
|
||
daily = _num(row.get('日销量预测'))
|
||
if not code or daily <= 0:
|
||
continue
|
||
profile = profiles.get(code) or {}
|
||
product_type = str(profile.get('产品类型') or '')
|
||
product_name = str(profile.get('货品名称') or '')
|
||
if product_type in ('包材', '原材料', '其他') or any(x in product_name for x in ('【内料】','【瓶子】','【瓶盖】','【泵头】','【彩盒】','【标签】')):
|
||
continue
|
||
lead = _int(profile.get('生产货期'))
|
||
logistics = _int(profile.get('物流时间'))
|
||
threshold = lead + logistics + (0 if product_type == '现货' else 7)
|
||
supply = sum(_int(row.get(k)) for k in ('库存', '加工厂库存', '生产在途', '采购在途'))
|
||
support = round(supply / daily, 1)
|
||
suggested = _int(row.get('建议下单数量'))
|
||
if support > horizon_days and support >= threshold and suggested <= 0:
|
||
continue
|
||
risk = _risk_level(support, threshold, daily)
|
||
shortage_day = date.today() + timedelta(days=max(0, int(math.floor(support))))
|
||
shortage_date = shortage_day.isoformat()
|
||
latest_order_date = (shortage_day - timedelta(days=threshold)).isoformat()
|
||
target = suggested or _round_qty(daily * threshold)
|
||
gaps = []
|
||
if target > 0:
|
||
for material in bom.get(code) or []:
|
||
supported = _int(material.get('可支撑成品数量'))
|
||
if _num(material.get('单位用量')) > 0 and supported < target:
|
||
gaps.append({'material_name': material.get('物料名称') or material.get('material_code') or '',
|
||
'support_qty': supported, 'gap': target - supported})
|
||
rows.append({
|
||
'product_code': code,
|
||
'product_name': product_name,
|
||
'brand': profile.get('品牌') or row.get('品牌') or '',
|
||
'daily_sales': round(daily, 2),
|
||
'supply_qty': supply,
|
||
'support_days': support,
|
||
'threshold_days': threshold,
|
||
'estimated_shortage_date': shortage_date,
|
||
'latest_order_date': latest_order_date,
|
||
'suggested_order_qty': suggested,
|
||
'risk_level': risk,
|
||
'bom_shortage_count': len(gaps),
|
||
'top_bom_gap': gaps[0] if gaps else None,
|
||
'planner': profile.get('采购计划') or '',
|
||
'follower': profile.get('采购跟单') or '',
|
||
'supplier': profile.get('供应商') or '',
|
||
})
|
||
rank = {'高': 3, '中': 2, '低': 1}
|
||
rows.sort(key=lambda x: (-rank.get(x['risk_level'], 0), x['support_days'], -x['suggested_order_qty']))
|
||
result = {
|
||
'generated_at': date.today().isoformat(),
|
||
'horizon_days': horizon_days,
|
||
'total': len(rows),
|
||
'items': rows[:limit],
|
||
'counts': {
|
||
'high': sum(1 for x in rows if x['risk_level'] == '高'),
|
||
'medium': sum(1 for x in rows if x['risk_level'] == '中'),
|
||
'low': sum(1 for x in rows if x['risk_level'] == '低'),
|
||
},
|
||
'data_dates': {
|
||
'inventory': (datasource_meta('product_inventory').get('data_date') or '') if 'product_inventory' in allowed else '',
|
||
'sales': (datasource_meta('product_sales').get('data_date') or '') if 'product_sales' in allowed else '',
|
||
},
|
||
}
|
||
cache_set(cache_key, result, ttl=300)
|
||
return result
|
||
|
||
|
||
def build_action_drafts(allowed_sources, profile, horizon_days=30, limit=30):
|
||
"""根据预测和主动洞察生成待人工确认的行动草稿，不执行任何写操作。"""
|
||
allowed = set(allowed_sources or [])
|
||
profile = profile or position_profile('')
|
||
drafts = []
|
||
if {'planning_metrics', 'product_profile'}.issubset(allowed):
|
||
forecast = forecast_supply_risks(horizon_days=horizon_days, limit=100, allowed_sources=allowed)
|
||
for item in forecast.get('items') or []:
|
||
if item['risk_level'] not in ('高', '中'):
|
||
continue
|
||
drafts.append({
|
||
'id': f"purchase-{item['product_code']}",
|
||
'action_type': '补货复核',
|
||
'priority': 'P0' if item['risk_level'] == '高' else 'P1',
|
||
'title': f"复核 {item['product_code']} 补货需求",
|
||
'product_code': item['product_code'],
|
||
'supplier': item.get('supplier') or '',
|
||
'suggested_qty': item.get('suggested_order_qty') or 0,
|
||
'owner': item.get('planner') or profile.get('name') or '采购计划',
|
||
'reason': f"预计{item['estimated_shortage_date']}库存见底，最晚应于{item.get('latest_order_date') or '尽快'}完成下单复核；支撑{item['support_days']}天，阈值{item['threshold_days']}天。",
|
||
'status': '待人工确认',
|
||
'source_page': '/planning',
|
||
'question': f"分析{item['product_code']}补货数量并检查BOM包材缺口",
|
||
})
|
||
|
||
raw = get_operational_insights()
|
||
insights = filter_insights(raw, allowed, profile, limit=100).get('items') or []
|
||
for item in insights:
|
||
if item.get('type') == 'overdue_order':
|
||
drafts.append({
|
||
'id': 'expedite-' + item['id'], 'action_type': '订单催办',
|
||
'priority': 'P0' if item['severity'] == '高' else 'P1',
|
||
'title': item['title'], 'product_code': item.get('product_code') or '',
|
||
'supplier': item.get('supplier') or '', 'suggested_qty': 0,
|
||
'owner': profile.get('name') or '采购跟单', 'reason': item['detail'],
|
||
'status': '待人工确认', 'source_page': item.get('source_page') or '/orders/domestic',
|
||
'question': item.get('question') or '',
|
||
})
|
||
elif item.get('type') == 'bom_shortage':
|
||
drafts.append({
|
||
'id': 'bom-' + item['id'], 'action_type': 'BOM补料复核',
|
||
'priority': 'P0' if item['severity'] == '高' else 'P1',
|
||
'title': item['title'], 'product_code': item.get('product_code') or '',
|
||
'supplier': item.get('supplier') or '',
|
||
'suggested_qty': _int((item.get('metrics') or {}).get('预计缺口')),
|
||
'owner': profile.get('name') or '采购计划', 'reason': item['detail'],
|
||
'status': '待人工确认', 'source_page': '/bom',
|
||
'question': item.get('question') or '',
|
||
})
|
||
# 去重、排序并限制数量。
|
||
unique = {}
|
||
for draft in drafts:
|
||
unique.setdefault(draft['id'], draft)
|
||
priority_rank = {'P0': 2, 'P1': 1, 'P2': 0}
|
||
rows = sorted(unique.values(), key=lambda x: (-priority_rank.get(x['priority'], 0), x['action_type'], x['title']))
|
||
return {
|
||
'generated_at': date.today().isoformat(),
|
||
'profile': profile,
|
||
'items': rows[:max(1, min(_int(limit) or 30, 100))],
|
||
'total': len(rows),
|
||
'notice': '所有行动均为只读草稿，必须由用户进入原业务菜单核对并确认后才能执行。',
|
||
}
|