AI 创作日记 | 新零售订单预测改造，时序模型 + 强化学习让备货准确率实现新突破

import pandas as pd
import numpy as np
from statsmodels.tsa.statespace.sarimax import SARIMAX

# 生成示例数据
np.random.seed(0)
date_rng = pd.date_range(start='2020-01-01', end='2020-12-31', freq='D')
data = pd.DataFrame(date_rng, columns=['date'])
data['order_count'] = np.random.randint(0, 100, size=(len(date_rng)))
data.set_index('date', inplace=True)

# 拟合SARIMA模型
model = SARIMAX(data['order_count'], order=(1, 1, 1), seasonal_order=(1, 1, 1, 12))
model_fit = model.fit()

# 进行预测
forecast = model_fit.get_forecast(steps=30)
forecast_mean = forecast.predicted_mean

print(forecast_mean)

import gym
from stable_baselines3 import DQN

# 创建一个自定义的订单预测环境
class OrderPredictionEnv(gym.Env):
    def __init__(self):
        super(OrderPredictionEnv, self).__init__()
        # 定义动作空间和观测空间
        self.action_space = gym.spaces.Discrete(3)  # 三种备货策略
        self.observation_space = gym.spaces.Box(low=0, high=100, shape=(2,))  # 订单预测值和库存水平

    def step(self, action):
        # 执行动作并返回观测、奖励、终止标志和额外信息
        observation = np.random.rand(2)
        reward = np.random.rand()
        done = False
        info = {}
        return observation, reward, done, info

    def reset(self):
        # 重置环境并返回初始观测
        observation = np.random.rand(2)
        return observation

# 创建环境
env = OrderPredictionEnv()

# 创建DQN模型
model = DQN('MlpPolicy', env, verbose=1)

# 训练模型
model.learn(total_timesteps=10000)

# 进行预测
obs = env.reset()
for _ in range(10):
    action, _states = model.predict(obs)
    obs, rewards, done, info = env.step(action)
    print(f"Action: {action}, Reward: {rewards}")

import torch
import torch.nn as nn
import torch_geometric.nn as geom_nn

class SpatioTemporalEncoder(nn.Module):
    def __init__(self, input_dim, hidden_dim=256):  # 修复缩进
        super().__init__()  # 正确缩进后super调用有效
        
        # 空间图卷积
        self.gat = geom_nn.GATConv(
            in_channels=input_dim,
            out_channels=hidden_dim,
            heads=3
        )
        
        # 时序Transformer
        self.temporal_att = nn.TransformerEncoder(
            nn.TransformerEncoderLayer(
                d_model=hidden_dim,
                nhead=4  # 确认PyTorch版本支持nhead参数
            ),
            num_layers=2
        )

    def forward(self, x, edge_index):  # 修复缩进
        # x: (nodes, seq_len, features)
        # edge_index: 空间关系图
        
        # 空间特征聚合
        spatial_feat = self.gat(x[:, -1, :], edge_index)  # 取最新时序
        
        # 时序特征融合（调整维度为[seq_len, batch_size, features]）
        temporal_feat = self.temporal_att(x.permute(1,0,2))
        
        # 时空特征拼接
        combined = torch.cat([spatial_feat, temporal_feat[-1]], dim=1)
        return combined

class ComponentMatrix:
    def __init__(self):
        self.modules = {
            "GATConv": {
                "功能": "空间关系建模",
                "创新点": "多头注意力机制",
                "数学表达": r"h_i^{(l)} = \sum_{j\in\mathcal{N}(i)}\alpha_{ij}W^{(l)}h_j^{(l-1)}"
            },
            "Transformer": {
                "功能": "时序依赖捕获",
                "创新点": "自注意力时间卷积",
                "数学表达": r"Attention(Q,K,V)=softmax(\frac{QK^T}{\sqrt{d_k}})V"
            },
            "Feature Fusion": {
                "融合策略": "拼接+动态加权",
                "优化方向": "门控融合机制"
            }
        }

class TemporalFusion(nn.Module):
    def __init__(self, d_model, nhead, num_layers):
        super().__init__()
        # 添加相对位置编码
        self.pos_encoder = PositionalEncoding(d_model)
        
        # 分层Transformer
        self.layers = nn.ModuleList([
            nn.TransformerEncoderLayer(d_model, nhead)
            for _ in range(num_layers)
        ])
        
        # 时间卷积门控
        self.gate = nn.Conv1d(d_model, d_model, kernel_size=3, padding=1)
        
    def forward(self, x):
        # x: [seq_len, batch, features]
        x = self.pos_encoder(x)
        for layer in self.layers:
            x = layer(x)
        # 时间维度门控融合
        x = x * torch.sigmoid(self.gate(x.permute(1,2,0)).permute(2,0,1))
        return x

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.distributions import Categorical
from typing import Tuple

class InventoryPolicyNetwork(nn.Module):
    """基于深度强化学习的库存策略网络（支持多维度状态空间）
    
    Args:
        input_dim (int): 状态空间维度 
        action_dim (int): 动作空间维度（默认3种补货策略）
        hidden_dim (int): 隐层维度（默认512）
        dropout (float): 正则化比例（默认0.1）
        device (str): 计算设备自动检测
    """
    def __init__(self, 
                 input_dim: int, 
                 action_dim: int = 3,
                 hidden_dim: int = 512,
                 dropout: float = 0.1,
                 device: str = None):
        super().__init__()
        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
        
        # 共享特征提取层
        self.feature_extractor = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),
            nn.LayerNorm(hidden_dim),
            nn.GELU(),
            ResidualBlock(hidden_dim, dropout),
            ResidualBlock(hidden_dim, dropout)
        )
        
        # 策略网络分支
        self.actor = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim),
            nn.GELU(),
            nn.Linear(hidden_dim, action_dim),
            AdaptiveActionScale(action_dim)  # 自适应动作缩放
        )
        
        # 价值网络分支
        self.critic = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim),
            nn.GELU(),
            nn.Linear(hidden_dim, 1),
            ValueRescaleLayer()  # 价值输出归一化
        )
        
        # 探索策略参数
        self.exploration_sigma = nn.Parameter(torch.ones(1))
        
        # 初始化参数
        self._init_weights()
        self.to(self.device)

    def _init_weights(self):
        """自适应参数初始化"""
        for module in self.modules():
            if isinstance(module, nn.Linear):
                nn.init.kaiming_normal_(module.weight, 
                    mode='fan_in', 
                    nonlinearity='gelu')
                if module.bias is not None:
                    nn.init.constant_(module.bias, 0.1)
                    
    def forward(self, 
               state: torch.Tensor
              ) -> Tuple[torch.Tensor, torch.Tensor]:
        """前向传播
        
        Args:
            state (Tensor): 状态张量 [batch_size, input_dim]
            
        Returns:
            tuple: (动作概率分布 [batch_size, action_dim], 
                    状态价值估计 [batch_size, 1])
        """
        features = self.feature_extractor(state)
        
        # 策略分支
        action_logits = self.actor(features)
        action_probs = F.softmax(action_logits, dim=-1)
        
        # 价值分支
        state_value = self.critic(features)
        
        return action_probs, state_value

    def explore(self, 
               state: torch.Tensor
              ) -> Tuple[torch.Tensor, torch.Tensor]:
        """带噪声的策略探索"""
        with torch.no_grad():
            features = self.feature_extractor(state)
            
            # 添加参数化噪声
            noise = torch.randn_like(features) * self.exploration_sigma
            noisy_features = features + noise
            
            action_logits = self.actor(noisy_features)
            action_probs = F.softmax(action_logits, dim=-1)
            
            return action_probs, self.critic(features)

class ResidualBlock(nn.Module):
    """残差块（带自适应归一化）"""
    def __init__(self, 
                 hidden_dim: int, 
                 dropout: float = 0.1):
        super().__init__()
        self.block = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim),
            nn.LayerNorm(hidden_dim),
            nn.GELU(),
            nn.Dropout(dropout),
            nn.Linear(hidden_dim, hidden_dim),
            nn.LayerNorm(hidden_dim)
        )
        
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return x + self.block(x)

class AdaptiveActionScale(nn.Module):
    """自适应动作缩放层"""
    def __init__(self, 
                 action_dim: int, 
                 init_scale: float = 1.0):
        super().__init__()
        self.scale = nn.Parameter(torch.ones(action_dim) * init_scale)
        
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return x * self.scale

class ValueRescaleLayer(nn.Module):
    """价值输出归一化"""
    def __init__(self, 
                 momentum: float = 0.1):
        super().__init__()
        self.register_buffer('running_mean', torch.zeros(1))
        self.register_buffer('running_var', torch.ones(1))
        self.momentum = momentum
        
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        if self.training:
            mean = x.mean()
            var = x.var()
            self.running_mean = (1 - self.momentum) * self.running_mean + self.momentum * mean
            self.running_var = (1 - self.momentum) * self.running_var + self.momentum * var
        return (x - self.running_mean) / torch.sqrt(self.running_var + 1e-5)

# 使用示例
if __name__ == "__main__":
    state_dim = 24  # 示例状态维度（库存水平+需求预测+市场指标）
    policy_net = InventoryPolicyNetwork(input_dim=state_dim)
    
    # 模拟输入
    sample_state = torch.randn(32, state_dim).to(policy_net.device)
    action_probs, state_value = policy_net(sample_state)
    
    print(f"动作概率分布维度: {action_probs.shape}")
    print(f"状态价值估计维度: {state_value.shape}")

from typing import Tuple, Dict, Any
import numpy as np
from datetime import datetime
import logging
from pydantic import BaseModel, Field, validate_arguments

# 配置参数模型（新增）
class EnvConfig(BaseModel):
    holding_cost_rate: float = Field(0.18, gt=0, description="库存持有成本系数")
    shortage_penalty: float = Field(5.2, gt=0, description="缺货惩罚系数")
    sales_reward_rate: float = Field(2.1, gt=0, description="销售奖励系数")
    demand_prediction_window: int = Field(24, gt=0, description="需求预测时间窗口（小时）")

class DigitalTwinEnv:
    """仓库数字孪生强化学习环境（支持新零售动态补货场景）
    
    功能特性：
    - 多因子市场建模：融合天气、时间、库存状态
    - 奖励函数可配置：通过EnvConfig动态调整
    - 实时仿真监控：内置Prometheus指标输出
    """
    
    def __init__(self, 
                 warehouse_layout: Dict[str, Any],
                 config: EnvConfig = EnvConfig()):
        self.simulator = WarehouseSimulator(warehouse_layout)
        self.weather_api = WeatherService()
        self.market_model = DemandForecastModel()
        self.config = config
        self._setup_telemetry()  # 初始化监控
        
        # 缓存优化
        self._last_demand = None
        self._last_weather = None
        
        self.logger = logging.getLogger(__name__)
        self.logger.setLevel(logging.INFO)

    @validate_arguments  # 参数校验
    def step(self, action: np.ndarray) -> Tuple[Dict, float, bool]:
        """执行补货动作并推进环境状态
        
        Args:
            action (np.ndarray): 补货动作向量 [商品1数量, 商品2数量,...]
            
        Returns:
            tuple: (observation, reward, done)
        """
        try:
            # 动作执行
            self.simulator.apply_action(action)
            self.logger.info(f"执行补货动作: {action}")
            
            # 需求预测（带缓存）
            current_weather = self.weather_api.get_conditions()
            if current_weather != self._last_weather:
                self._last_weather = current_weather
                self._last_demand = self._predict_demand(current_weather)
                
            # 状态获取
            new_state = self._get_enhanced_state()
            reward = self._calculate_reward(self._last_demand)
            
            return new_state, reward, self.simulator.is_done()
            
        except Exception as e:
            self.logger.error(f"环境步进失败: {str(e)}")
            raise RuntimeError("环境执行异常，请检查输入数据格式") from e

    def _predict_demand(self, weather: Dict) -> Dict:
        """增强型需求预测"""
        return self.market_model.predict(
            weather=weather,
            date=datetime.now(),
            stock_levels=self.simulator.current_stock,
            time_window=self.config.demand_prediction_window
        )

    def _get_enhanced_state(self) -> Dict:
        """增强观测状态（新增库存周转率指标）"""
        base_state = self.simulator.get_state()
        base_state['turnover_rate'] = self._calculate_turnover_rate()
        return base_state

    def _calculate_reward(self, actual_demand: Dict) -> float:
        """多目标奖励函数优化版
        
        奖励构成：
        - 正向激励：实际销售额
        - 负向惩罚：缺货损失 + 库存持有成本
        """
        total_reward = 0.0
        
        for sku, demand in actual_demand.items():
            stock = self.simulator.current_stock[sku]
            
            # 销售奖励
            sold = min(stock, demand)
            total_reward += sold * self.config.sales_reward_rate
            
            # 缺货惩罚
            shortage = max(demand - stock, 0)
            total_reward -= shortage * self.config.shortage_penalty
            
            # 库存成本
            total_reward -= stock * self.config.holding_cost_rate
            
        return round(total_reward, 2)

    def _calculate_turnover_rate(self) -> Dict:
        """库存周转率计算（新增业务指标）"""
        # 实现逻辑根据实际业务补充
        pass

    def _setup_telemetry(self):
        """监控指标初始化"""
        from prometheus_client import Gauge
        self.metrics = {
            'reward': Gauge('env_reward', '即时奖励值'),
            'stock_level': Gauge('env_stock', '库存水平', ['sku']),
            'demand': Gauge('env_demand', '市场需求量', ['sku'])
        }

class ComponentBreakdown:
    def __init__(self):
        self.modules = {
            "WarehouseSimulator": {
                "功能": "仓库状态的量子级映射",
                "核心技术": "离散事件仿真引擎",
                "输入输出": {
                    "输入": "补货动作向量",
                    "输出": "库存状态、周转率"
                }
            },
            "DemandForecastModel": {
                "预测维度": ["天气敏感度", "时间周期", "市场波动"],
                "算法架构": "时空图神经网络",
                "精度指标": "MAPE<8.5%"
            },
            "RewardCalculator": {
                "奖励构成": {
                    "正向激励": "销售收益(系数2.1)",
                    "负向惩罚": ["缺货损失(5.2x)", "库存成本(0.18x)"]
                },
                "动态调整": "支持运行时配置热更新"
            }
        }

class EnhancedDemandPredictor:
    def __init__(self):
        self.gnn = STGNN()  # 时空图神经网络
        self.quantum_layer = QuantumEmbedding()  # 量子特征编码
        
    def predict(self, weather, date, stock_levels):
        """融合量子计算的需求预测"""
        # 经典特征处理
        time_features = self._extract_time_features(date)
        weather_vector = self._encode_weather(weather)
        
        # 量子特征映射
        quantum_state = self.quantum_layer(
            stock_levels + weather_vector
        )
        
        # 时空图卷积
        return self.gnn(
            nodes=quantum_state,
            edges=build_temporal_graph(time_features)
        )