{"id":65786,"date":"2026-01-25T17:40:35","date_gmt":"2026-01-25T09:40:35","guid":{"rendered":"https:\/\/www.wsisp.com\/helps\/65786.html"},"modified":"2026-01-25T17:40:35","modified_gmt":"2026-01-25T09:40:35","slug":"%e7%b1%b3%e5%93%88%e6%b8%b8java%e9%9d%a2%e8%af%95%e8%a2%ab%e9%97%ae%ef%bc%9a%e6%9c%ba%e5%99%a8%e5%ad%a6%e4%b9%a0%e6%a8%a1%e5%9e%8b%e7%9a%84%e5%9c%a8%e7%ba%bf%e6%9c%8d%e5%8a%a1%e5%92%8ca-b%e6%b5%8b","status":"publish","type":"post","link":"https:\/\/www.wsisp.com\/helps\/65786.html","title":{"rendered":"\u7c73\u54c8\u6e38Java\u9762\u8bd5\u88ab\u95ee\uff1a\u673a\u5668\u5b66\u4e60\u6a21\u578b\u7684\u5728\u7ebf\u670d\u52a1\u548cA\/B\u6d4b\u8bd5"},"content":{"rendered":"

\u4e00\u3001\u673a\u5668\u5b66\u4e60\u5728\u7ebf\u670d\u52a1\u6838\u5fc3\u67b6\u6784<\/h3>\n

1.1 \u7edf\u4e00\u6a21\u578b\u670d\u52a1\u67b6\u6784<\/h4>\n

python<\/p>\n<\/p>\n

\u590d\u5236<\/p>\n<\/p>\n

\u4e0b\u8f7d<\/p>\n

"""
\n\u673a\u5668\u5b66\u4e60\u6a21\u578b\u5728\u7ebf\u670d\u52a1\u67b6\u6784
\n"""
\nfrom abc import ABC, abstractmethod
\nfrom typing import Dict, Any, List, Optional, Union
\nfrom dataclasses import dataclass, asdict
\nimport json
\nimport time
\nimport asyncio
\nimport threading
\nfrom concurrent.futures import ThreadPoolExecutor
\nfrom datetime import datetime
\nfrom enum import Enum
\nimport pickle
\nimport numpy as np
\nimport pandas as pd
\nfrom pydantic import BaseModel, Field, validator
\nimport logging
\nfrom prometheus_client import Counter, Histogram, Gauge
\nfrom contextlib import contextmanager<\/p>\n

# ==================== \u76d1\u63a7\u6307\u6807 ====================
\nclass ModelMetrics:
\n """\u6a21\u578b\u6027\u80fd\u76d1\u63a7\u6307\u6807"""<\/p>\n

# \u8bf7\u6c42\u76f8\u5173\u6307\u6807
\n request_total = Counter('model_request_total', 'Total requests', ['model_name', 'version'])
\n request_duration = Histogram('model_request_duration_seconds', 'Request duration', ['model_name', 'version'])
\n request_errors = Counter('model_request_errors', 'Request errors', ['model_name', 'version', 'error_type'])<\/p>\n

# \u9884\u6d4b\u76f8\u5173\u6307\u6807
\n prediction_latency = Histogram('prediction_latency_seconds', 'Prediction latency', ['model_name', 'version'])
\n prediction_accuracy = Gauge('prediction_accuracy', 'Prediction accuracy', ['model_name', 'version'])<\/p>\n

# \u8d44\u6e90\u76f8\u5173\u6307\u6807
\n model_memory_usage = Gauge('model_memory_usage_bytes', 'Model memory usage', ['model_name', 'version'])
\n model_load_time = Gauge('model_load_time_seconds', 'Model load time', ['model_name', 'version'])<\/p>\n

@dataclass
\nclass ModelMetadata:
\n """\u6a21\u578b\u5143\u6570\u636e"""
\n name: str
\n version: str
\n created_at: datetime
\n framework: str # tensorflow, pytorch, sklearn, xgboost\u7b49
\n input_schema: Dict[str, Any] # \u8f93\u5165\u6570\u636e\u6a21\u5f0f
\n output_schema: Dict[str, Any] # \u8f93\u51fa\u6570\u636e\u6a21\u5f0f
\n performance_metrics: Dict[str, float] # \u6027\u80fd\u6307\u6807
\n feature_importance: Optional[List[float]] = None
\n dependencies: Optional[List[str]] = None
\n description: Optional[str] = None<\/p>\n

def to_dict(self) -> Dict[str, Any]:
\n return asdict(self)<\/p>\n

class ModelStatus(Enum):
\n """\u6a21\u578b\u72b6\u6001\u679a\u4e3e"""
\n LOADING = "loading"
\n READY = "ready"
\n UNHEALTHY = "unhealthy"
\n OFFLINE = "offline"<\/p>\n

class ModelType(Enum):
\n """\u6a21\u578b\u7c7b\u578b\u679a\u4e3e"""
\n CLASSIFICATION = "classification"
\n REGRESSION = "regression"
\n CLUSTERING = "clustering"
\n RECOMMENDATION = "recommendation"
\n NLP = "nlp"
\n CV = "computer_vision"<\/p>\n

# ==================== \u57fa\u7840\u6a21\u578b\u63a5\u53e3 ====================
\nclass BaseModelInterface(ABC):
\n """\u57fa\u7840\u6a21\u578b\u63a5\u53e3"""<\/p>\n

def __init__(self, model_name: str, model_version: str):
\n self.model_name = model_name
\n self.model_version = model_version
\n self.metadata: Optional[ModelMetadata] = None
\n self.status: ModelStatus = ModelStatus.LOADING
\n self.loaded_at: Optional[datetime] = None
\n self.metrics = ModelMetrics()
\n self.logger = logging.getLogger(f"{model_name}.{model_version}")<\/p>\n

@abstractmethod
\n async def load(self, model_path: str) -> bool:
\n """\u52a0\u8f7d\u6a21\u578b"""
\n pass<\/p>\n

@abstractmethod
\n async def predict(self, inputs: Union[Dict, List, np.ndarray]) -> Dict[str, Any]:
\n """\u6a21\u578b\u9884\u6d4b"""
\n pass<\/p>\n

@abstractmethod
\n async def batch_predict(self, inputs: List) -> List[Dict[str, Any]]:
\n """\u6279\u91cf\u9884\u6d4b"""
\n pass<\/p>\n

@abstractmethod
\n def get_feature_names(self) -> List[str]:
\n """\u83b7\u53d6\u7279\u5f81\u540d\u79f0"""
\n pass<\/p>\n

async def health_check(self) -> Dict[str, Any]:
\n """\u5065\u5eb7\u68c0\u67e5"""
\n return {
\n "status": self.status.value,
\n "model_name": self.model_name,
\n "model_version": self.model_version,
\n "loaded_at": self.loaded_at.isoformat() if self.loaded_at else None,
\n "memory_usage": self._get_memory_usage(),
\n "uptime": self._get_uptime()
\n }<\/p>\n

def _get_memory_usage(self) -> Optional[int]:
\n """\u83b7\u53d6\u5185\u5b58\u4f7f\u7528\u91cf"""
\n # \u5177\u4f53\u5b9e\u73b0\u53d6\u51b3\u4e8e\u6846\u67b6
\n return None<\/p>\n

def _get_uptime(self) -> Optional[float]:
\n """\u83b7\u53d6\u8fd0\u884c\u65f6\u95f4"""
\n if self.loaded_at:
\n return (datetime.now() – self.loaded_at).total_seconds()
\n return None<\/p>\n

@contextmanager
\n def track_request(self):
\n """\u8bf7\u6c42\u8ddf\u8e2a\u4e0a\u4e0b\u6587\u7ba1\u7406\u5668"""
\n start_time = time.time()
\n try:
\n self.metrics.request_total.labels(
\n model_name=self.model_name,
\n version=self.model_version
\n ).inc()
\n yield
\n except Exception as e:
\n self.metrics.request_errors.labels(
\n model_name=self.model_name,
\n version=self.model_version,
\n error_type=type(e).__name__
\n ).inc()
\n raise
\n finally:
\n duration = time.time() – start_time
\n self.metrics.request_duration.labels(
\n model_name=self.model_name,
\n version=self.model_version
\n ).observe(duration)<\/p>\n

# ==================== \u5177\u4f53\u6a21\u578b\u5b9e\u73b0 ====================
\nclass ScikitLearnModel(BaseModelInterface):
\n """Scikit-learn\u6a21\u578b\u5b9e\u73b0"""<\/p>\n

def __init__(self, model_name: str, model_version: str):
\n super().__init__(model_name, model_version)
\n self.model = None
\n self.feature_names: List[str] = []<\/p>\n

async def load(self, model_path: str) -> bool:
\n """\u52a0\u8f7d\u6a21\u578b"""
\n with self.track_request():
\n try:
\n with open(model_path, 'rb') as f:
\n data = pickle.load(f)<\/p>\n

self.model = data['model']
\n self.feature_names = data.get('feature_names', [])
\n self.metadata = data.get('metadata')<\/p>\n

self.status = ModelStatus.READY
\n self.loaded_at = datetime.now()<\/p>\n

# \u8bb0\u5f55\u52a0\u8f7d\u65f6\u95f4
\n self.metrics.model_load_time.labels(
\n model_name=self.model_name,
\n version=self.model_version
\n ).set(time.time() – self.loaded_at.timestamp())<\/p>\n

self.logger.info(f"\u6a21\u578b\u52a0\u8f7d\u6210\u529f: {self.model_name}:{self.model_version}")
\n return True<\/p>\n

except Exception as e:
\n self.status = ModelStatus.UNHEALTHY
\n self.logger.error(f"\u6a21\u578b\u52a0\u8f7d\u5931\u8d25: {e}")
\n return False<\/p>\n

async def predict(self, inputs: Union[Dict, List, np.ndarray]) -> Dict[str, Any]:
\n """\u5355\u6761\u9884\u6d4b"""
\n with self.track_request():
\n try:
\n start_time = time.time()<\/p>\n

# \u8f93\u5165\u9884\u5904\u7406
\n processed_inputs = self._preprocess(inputs)<\/p>\n

# \u6267\u884c\u9884\u6d4b
\n prediction = self.model.predict(processed_inputs)
\n probabilities = None<\/p>\n

# \u5982\u679c\u6a21\u578b\u652f\u6301\u6982\u7387\u9884\u6d4b
\n if hasattr(self.model, 'predict_proba'):
\n probabilities = self.model.predict_proba(processed_inputs)<\/p>\n

# \u8bb0\u5f55\u9884\u6d4b\u5ef6\u8fdf
\n latency = time.time() – start_time
\n self.metrics.prediction_latency.labels(
\n model_name=self.model_name,
\n version=self.model_version
\n ).observe(latency)<\/p>\n

return {
\n "prediction": prediction.tolist() if hasattr(prediction, 'tolist') else prediction,
\n "probabilities": probabilities.tolist() if probabilities is not None else None,
\n "model_name": self.model_name,
\n "model_version": self.model_version,
\n "latency_ms": latency * 1000,
\n "timestamp": datetime.now().isoformat()
\n }<\/p>\n

except Exception as e:
\n self.logger.error(f"\u9884\u6d4b\u5931\u8d25: {e}")
\n raise<\/p>\n

async def batch_predict(self, inputs: List) -> List[Dict[str, Any]]:
\n """\u6279\u91cf\u9884\u6d4b"""
\n results = []
\n for input_data in inputs:
\n try:
\n result = await self.predict(input_data)
\n results.append(result)
\n except Exception as e:
\n results.append({
\n "error": str(e),
\n "model_name": self.model_name,
\n "model_version": self.model_version,
\n "timestamp": datetime.now().isoformat()
\n })
\n return results<\/p>\n

def get_feature_names(self) -> List[str]:
\n """\u83b7\u53d6\u7279\u5f81\u540d\u79f0"""
\n return self.feature_names<\/p>\n

def _preprocess(self, inputs: Union[Dict, List, np.ndarray]) -> np.ndarray:
\n """\u8f93\u5165\u6570\u636e\u9884\u5904\u7406"""
\n if isinstance(inputs, dict):
\n # \u5b57\u5178\u8f6c\u6570\u7ec4，\u6309\u7279\u5f81\u540d\u79f0\u6392\u5e8f
\n if self.feature_names:
\n return np.array([inputs.get(feat, 0) for feat in self.feature_names]).reshape(1, -1)
\n else:
\n return np.array(list(inputs.values())).reshape(1, -1)
\n elif isinstance(inputs, list):
\n return np.array(inputs).reshape(1, -1)
\n elif isinstance(inputs, np.ndarray):
\n return inputs.reshape(1, -1) if inputs.ndim == 1 else inputs
\n else:
\n raise ValueError(f"\u4e0d\u652f\u6301\u7684\u8f93\u5165\u7c7b\u578b: {type(inputs)}")<\/p>\n

class TensorFlowModel(BaseModelInterface):
\n """TensorFlow\u6a21\u578b\u5b9e\u73b0"""<\/p>\n

def __init__(self, model_name: str, model_version: str):
\n super().__init__(model_name, model_version)
\n try:
\n import tensorflow as tf
\n self.tf = tf
\n self.model = None
\n except ImportError:
\n raise ImportError("TensorFlow\u672a\u5b89\u88c5")<\/p>\n

async def load(self, model_path: str) -> bool:
\n """\u52a0\u8f7dTensorFlow\u6a21\u578b"""
\n with self.track_request():
\n try:
\n self.model = self.tf.keras.models.load_model(model_path)<\/p>\n

# \u52a0\u8f7d\u5143\u6570\u636e
\n metadata_path = model_path.replace('.h5', '_metadata.json').replace('.keras', '_metadata.json')
\n if os.path.exists(metadata_path):
\n with open(metadata_path, 'r') as f:
\n metadata_data = json.load(f)
\n self.metadata = ModelMetadata(**metadata_data)<\/p>\n

self.status = ModelStatus.READY
\n self.loaded_at = datetime.now()<\/p>\n

self.logger.info(f"TensorFlow\u6a21\u578b\u52a0\u8f7d\u6210\u529f: {self.model_name}:{self.model_version}")
\n return True<\/p>\n

except Exception as e:
\n self.status = ModelStatus.UNHEALTHY
\n self.logger.error(f"TensorFlow\u6a21\u578b\u52a0\u8f7d\u5931\u8d25: {e}")
\n return False<\/p>\n

async def predict(self, inputs: Union[Dict, List, np.ndarray]) -> Dict[str, Any]:
\n """TensorFlow\u6a21\u578b\u9884\u6d4b"""
\n with self.track_request():
\n try:
\n start_time = time.time()<\/p>\n

# \u8f93\u5165\u9884\u5904\u7406
\n processed_inputs = self._preprocess(inputs)<\/p>\n

# \u6267\u884c\u9884\u6d4b
\n prediction = self.model.predict(processed_inputs, verbose=0)<\/p>\n

# \u8bb0\u5f55\u9884\u6d4b\u5ef6\u8fdf
\n latency = time.time() – start_time
\n self.metrics.prediction_latency.labels(
\n model_name=self.model_name,
\n version=self.model_version
\n ).observe(latency)<\/p>\n

return {
\n "prediction": prediction.tolist(),
\n "model_name": self.model_name,
\n "model_version": self.model_version,
\n "latency_ms": latency * 1000,
\n "timestamp": datetime.now().isoformat()
\n }<\/p>\n

except Exception as e:
\n self.logger.error(f"TensorFlow\u9884\u6d4b\u5931\u8d25: {e}")
\n raise<\/p>\n

# … \u5176\u4ed6\u65b9\u6cd5\u5b9e\u73b0\u7c7b\u4f3cScikitLearnModel<\/p>\n

class PyTorchModel(BaseModelInterface):
\n """PyTorch\u6a21\u578b\u5b9e\u73b0"""<\/p>\n

def __init__(self, model_name: str, model_version: str):
\n super().__init__(model_name, model_version)
\n try:
\n import torch
\n self.torch = torch
\n self.model = None
\n self.device = self.torch.device('cuda' if self.torch.cuda.is_available() else 'cpu')
\n except ImportError:
\n raise ImportError("PyTorch\u672a\u5b89\u88c5")<\/p>\n

async def load(self, model_path: str) -> bool:
\n """\u52a0\u8f7dPyTorch\u6a21\u578b"""
\n with self.track_request():
\n try:
\n self.model = self.torch.load(model_path, map_location=self.device)
\n self.model.to(self.device)
\n self.model.eval()<\/p>\n

# \u52a0\u8f7d\u5143\u6570\u636e
\n metadata_path = model_path.replace('.pt', '_metadata.json').replace('.pth', '_metadata.json')
\n if os.path.exists(metadata_path):
\n with open(metadata_path, 'r') as f:
\n metadata_data = json.load(f)
\n self.metadata = ModelMetadata(**metadata_data)<\/p>\n

self.status = ModelStatus.READY
\n self.loaded_at = datetime.now()<\/p>\n

self.logger.info(f"PyTorch\u6a21\u578b\u52a0\u8f7d\u6210\u529f: {self.model_name}:{self.model_version}")
\n return True<\/p>\n

except Exception as e:
\n self.status = ModelStatus.UNHEALTHY
\n self.logger.error(f"PyTorch\u6a21\u578b\u52a0\u8f7d\u5931\u8d25: {e}")
\n return False<\/p>\n

async def predict(self, inputs: Union[Dict, List, np.ndarray]) -> Dict[str, Any]:
\n """PyTorch\u6a21\u578b\u9884\u6d4b"""
\n with self.track_request():
\n try:
\n start_time = time.time()<\/p>\n

# \u8f93\u5165\u9884\u5904\u7406
\n processed_inputs = self._preprocess(inputs)<\/p>\n

# \u8f6c\u6362\u4e3aTensor
\n inputs_tensor = self.torch.from_numpy(processed_inputs).float().to(self.device)<\/p>\n

# \u6267\u884c\u9884\u6d4b（\u4e0d\u8ba1\u7b97\u68af\u5ea6）
\n with self.torch.no_grad():
\n prediction = self.model(inputs_tensor)<\/p>\n

# \u8f6c\u6362\u56denumpy
\n prediction_np = prediction.cpu().numpy()<\/p>\n

# \u8bb0\u5f55\u9884\u6d4b\u5ef6\u8fdf
\n latency = time.time() – start_time
\n self.metrics.prediction_latency.labels(
\n model_name=self.model_name,
\n version=self.model_version
\n ).observe(latency)<\/p>\n

return {
\n "prediction": prediction_np.tolist(),
\n "model_name": self.model_name,
\n "model_version": self.model_version,
\n "latency_ms": latency * 1000,
\n "timestamp": datetime.now().isoformat()
\n }<\/p>\n

except Exception as e:
\n self.logger.error(f"PyTorch\u9884\u6d4b\u5931\u8d25: {e}")
\n raise<\/p>\n

# … \u5176\u4ed6\u65b9\u6cd5\u5b9e\u73b0\u7c7b\u4f3cScikitLearnModel<\/p>\n

# ==================== \u6a21\u578b\u5de5\u5382 ====================
\nclass ModelFactory:
\n """\u6a21\u578b\u5de5\u5382"""<\/p>\n

_model_registry = {
\n 'sklearn': ScikitLearnModel,
\n 'tensorflow': TensorFlowModel,
\n 'pytorch': PyTorchModel,
\n 'xgboost': ScikitLearnModel, # \u590d\u7528sklearn\u63a5\u53e3
\n 'lightgbm': ScikitLearnModel,
\n }<\/p>\n

@classmethod
\n def create_model(cls, framework: str, model_name: str, model_version: str) -> BaseModelInterface:
\n """\u521b\u5efa\u6a21\u578b\u5b9e\u4f8b"""
\n if framework not in cls._model_registry:
\n raise ValueError(f"\u4e0d\u652f\u6301\u7684\u6846\u67b6: {framework}")<\/p>\n

model_class = cls._model_registry[framework]
\n return model_class(model_name, model_version)<\/p>\n

@classmethod
\n def register_framework(cls, framework: str, model_class):
\n """\u6ce8\u518c\u65b0\u7684\u6a21\u578b\u6846\u67b6"""
\n if not issubclass(model_class, BaseModelInterface):
\n raise TypeError("\u6a21\u578b\u7c7b\u5fc5\u987b\u7ee7\u627f\u81eaBaseModelInterface")
\n cls._model_registry[framework] = model_class<\/p>\n

# ==================== \u6a21\u578b\u7ba1\u7406\u5668 ====================
\nclass ModelManager:
\n """\u6a21\u578b\u7ba1\u7406\u5668"""<\/p>\n

def __init__(self, config: Dict[str, Any]):
\n self.config = config
\n self.models: Dict[str, BaseModelInterface] = {}
\n self.model_registry: Dict[str, Dict[str, Any]] = {}
\n self.logger = logging.getLogger("ModelManager")
\n self.executor = ThreadPoolExecutor(max_workers=config.get('max_workers', 10))<\/p>\n

# \u521d\u59cb\u5316\u6a21\u578b\u6ce8\u518c\u8868
\n self._init_model_registry()<\/p>\n

def _init_model_registry(self):
\n """\u521d\u59cb\u5316\u6a21\u578b\u6ce8\u518c\u8868"""
\n # \u53ef\u4ee5\u4ece\u6570\u636e\u5e93\u6216\u914d\u7f6e\u6587\u4ef6\u4e2d\u52a0\u8f7d
\n default_models = self.config.get('models', [])
\n for model_config in default_models:
\n model_key = f"{model_config['name']}:{model_config['version']}"
\n self.model_registry[model_key] = model_config<\/p>\n

async def load_model(self, model_name: str, model_version: str) -> bool:
\n """\u52a0\u8f7d\u6a21\u578b"""
\n model_key = f"{model_name}:{model_version}"<\/p>\n

if model_key in self.models:
\n self.logger.info(f"\u6a21\u578b\u5df2\u52a0\u8f7d: {model_key}")
\n return True<\/p>\n

if model_key not in self.model_registry:
\n self.logger.error(f"\u6a21\u578b\u672a\u6ce8\u518c: {model_key}")
\n return False<\/p>\n

model_config = self.model_registry[model_key]
\n framework = model_config.get('framework', 'sklearn')
\n model_path = model_config.get('model_path')<\/p>\n

if not model_path:
\n self.logger.error(f"\u6a21\u578b\u8def\u5f84\u672a\u914d\u7f6e: {model_key}")
\n return False<\/p>\n

try:
\n # \u521b\u5efa\u6a21\u578b\u5b9e\u4f8b
\n model = ModelFactory.create_model(framework, model_name, model_version)<\/p>\n

# \u52a0\u8f7d\u6a21\u578b
\n success = await model.load(model_path)<\/p>\n

if success:
\n self.models[model_key] = model
\n self.logger.info(f"\u6a21\u578b\u52a0\u8f7d\u6210\u529f: {model_key}")
\n return True
\n else:
\n self.logger.error(f"\u6a21\u578b\u52a0\u8f7d\u5931\u8d25: {model_key}")
\n return False<\/p>\n

except Exception as e:
\n self.logger.error(f"\u6a21\u578b\u52a0\u8f7d\u5f02\u5e38: {model_key}, \u9519\u8bef: {e}")
\n return False<\/p>\n

async def unload_model(self, model_name: str, model_version: str) -> bool:
\n """\u5378\u8f7d\u6a21\u578b"""
\n model_key = f"{model_name}:{model_version}"<\/p>\n

if model_key not in self.models:
\n self.logger.warning(f"\u6a21\u578b\u672a\u52a0\u8f7d: {model_key}")
\n return False<\/p>\n

# \u6e05\u7406\u6a21\u578b\u8d44\u6e90
\n del self.models[model_key]<\/p>\n

# \u89e6\u53d1\u5783\u573e\u56de\u6536
\n import gc
\n gc.collect()<\/p>\n

self.logger.info(f"\u6a21\u578b\u5378\u8f7d\u6210\u529f: {model_key}")
\n return True<\/p>\n

async def predict(
\n self,
\n model_name: str,
\n model_version: str,
\n inputs: Union[Dict, List, np.ndarray]
\n ) -> Dict[str, Any]:
\n """\u6a21\u578b\u9884\u6d4b"""
\n model_key = f"{model_name}:{model_version}"<\/p>\n

if model_key not in self.models:
\n # \u5c1d\u8bd5\u52a0\u8f7d\u6a21\u578b
\n loaded = await self.load_model(model_name, model_version)
\n if not loaded:
\n raise ValueError(f"\u6a21\u578b\u672a\u52a0\u8f7d\u4e14\u52a0\u8f7d\u5931\u8d25: {model_key}")<\/p>\n

model = self.models[model_key]
\n return await model.predict(inputs)<\/p>\n

async def batch_predict(
\n self,
\n model_name: str,
\n model_version: str,
\n inputs: List
\n ) -> List[Dict[str, Any]]:
\n """\u6279\u91cf\u9884\u6d4b"""
\n model_key = f"{model_name}:{model_version}"<\/p>\n

if model_key not in self.models:
\n loaded = await self.load_model(model_name, model_version)
\n if not loaded:
\n raise ValueError(f"\u6a21\u578b\u672a\u52a0\u8f7d\u4e14\u52a0\u8f7d\u5931\u8d25: {model_key}")<\/p>\n

model = self.models[model_key]
\n return await model.batch_predict(inputs)<\/p>\n

def list_models(self) -> List[Dict[str, Any]]:
\n """\u5217\u51fa\u6240\u6709\u6a21\u578b"""
\n result = []
\n for model_key, model in self.models.items():
\n result.append({
\n 'model_key': model_key,
\n 'status': model.status.value,
\n 'loaded_at': model.loaded_at.isoformat() if model.loaded_at else None,
\n 'metadata': model.metadata.to_dict() if model.metadata else None
\n })
\n return result<\/p>\n

async def health_check_all(self) -> Dict[str, Any]:
\n """\u68c0\u67e5\u6240\u6709\u6a21\u578b\u5065\u5eb7\u72b6\u6001"""
\n health_status = {}<\/p>\n

for model_key, model in self.models.items():
\n try:
\n health = await model.health_check()
\n health_status[model_key] = health
\n except Exception as e:
\n health_status[model_key] = {
\n 'status': 'error',
\n 'error': str(e)
\n }<\/p>\n

return {
\n 'timestamp': datetime.now().isoformat(),
\n 'total_models': len(self.models),
\n 'healthy_models': sum(1 for h in health_status.values()
\n if h.get('status') == 'ready'),
\n 'details': health_status
\n }<\/p>\n

# ==================== \u7279\u5f81\u5de5\u7a0b\u670d\u52a1 ====================
\nclass FeatureEngineeringService:
\n """\u7279\u5f81\u5de5\u7a0b\u670d\u52a1"""<\/p>\n

def __init__(self):
\n self.feature_pipelines: Dict[str, Any] = {}
\n self.feature_store = {} # \u7b80\u5355\u7684\u5185\u5b58\u7279\u5f81\u5b58\u50a8，\u751f\u4ea7\u73af\u5883\u5e94\u4f7f\u7528Redis\u7b49
\n self.logger = logging.getLogger("FeatureEngineeringService")<\/p>\n

async def transform(self, pipeline_name: str, raw_data: Dict[str, Any]) -> Dict[str, Any]:
\n """\u7279\u5f81\u8f6c\u6362"""
\n if pipeline_name not in self.feature_pipelines:
\n raise ValueError(f"\u7279\u5f81\u7ba1\u9053\u672a\u627e\u5230: {pipeline_name}")<\/p>\n

pipeline = self.feature_pipelines[pipeline_name]<\/p>\n

try:
\n # \u6267\u884c\u7279\u5f81\u8f6c\u6362
\n transformed_features = pipeline.transform(raw_data)<\/p>\n

# \u8bb0\u5f55\u7279\u5f81\u4f7f\u7528
\n await self._log_feature_usage(pipeline_name, transformed_features)<\/p>\n

return transformed_features<\/p>\n

except Exception as e:
\n self.logger.error(f"\u7279\u5f81\u8f6c\u6362\u5931\u8d25: {pipeline_name}, \u9519\u8bef: {e}")
\n raise<\/p>\n

async def batch_transform(self, pipeline_name: str, raw_data_list: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
\n """\u6279\u91cf\u7279\u5f81\u8f6c\u6362"""
\n results = []
\n for raw_data in raw_data_list:
\n try:
\n transformed = await self.transform(pipeline_name, raw_data)
\n results.append(transformed)
\n except Exception as e:
\n results.append({"error": str(e)})
\n return results<\/p>\n

def register_pipeline(self, pipeline_name: str, pipeline: Any):
\n """\u6ce8\u518c\u7279\u5f81\u7ba1\u9053"""
\n self.feature_pipelines[pipeline_name] = pipeline
\n self.logger.info(f"\u7279\u5f81\u7ba1\u9053\u6ce8\u518c\u6210\u529f: {pipeline_name}")<\/p>\n

async def _log_feature_usage(self, pipeline_name: str, features: Dict[str, Any]):
\n """\u8bb0\u5f55\u7279\u5f81\u4f7f\u7528\u60c5\u51b5"""
\n # \u8fd9\u91cc\u53ef\u4ee5\u8bb0\u5f55\u5230\u76d1\u63a7\u7cfb\u7edf\u6216\u6570\u636e\u5e93
\n feature_stats = {
\n 'pipeline_name': pipeline_name,
\n 'feature_count': len(features),
\n 'timestamp': datetime.now().isoformat(),
\n 'features': list(features.keys())
\n }<\/p>\n

# \u7b80\u5355\u5185\u5b58\u5b58\u50a8，\u751f\u4ea7\u73af\u5883\u5e94\u4f7f\u7528\u6301\u4e45\u5316\u5b58\u50a8
\n key = f"feature_usage:{datetime.now().strftime('%Y%m%d')}:{pipeline_name}"
\n if key not in self.feature_store:
\n self.feature_store[key] = []
\n self.feature_store[key].append(feature_stats) <\/p>\n

\u7bc7\u5e45\u9650\u5236\u4e0b\u9762\u5c31\u53ea\u80fd\u7ed9\u5927\u5bb6\u5c55\u793a\u5c0f\u518c\u90e8\u5206\u5185\u5bb9\u4e86\u3002\u6574\u7406\u4e86\u4e00\u4efd\u6838\u5fc3\u9762\u8bd5\u7b14\u8bb0\u5305\u62ec\u4e86：Java\u9762\u8bd5\u3001Spring\u3001JVM\u3001MyBatis\u3001Redis\u3001MySQL\u3001\u5e76\u53d1\u7f16\u7a0b\u3001\u5fae\u670d\u52a1\u3001Linux\u3001Springboot\u3001SpringCloud\u3001MQ\u3001Kafc<\/p>\n

\u9700\u8981\u5168\u5957\u9762\u8bd5\u7b14\u8bb0\u53ca\u7b54\u6848 \u3010\u70b9\u51fb\u6b64\u5904\u5373\u53ef\/\u514d\u8d39\u83b7\u53d6\u3011\u200b\u200b\u200b<\/span><\/p>\n<\/p>\n

1.2 \u9ad8\u6027\u80fd\u6a21\u578b\u670d\u52a1API<\/h4>\n

python<\/p>\n<\/p>\n

\u590d\u5236<\/p>\n<\/p>\n

\u4e0b\u8f7d<\/p>\n

"""
\nFastAPI\u6a21\u578b\u670d\u52a1API
\n"""
\nfrom fastapi import FastAPI, HTTPException, Depends, BackgroundTasks, Query, Header
\nfrom fastapi.middleware.cors import CORSMiddleware
\nfrom fastapi.responses import JSONResponse
\nimport uvicorn
\nfrom typing import List, Optional
\nimport json
\nimport asyncio
\nfrom datetime import datetime<\/p>\n

# Pydantic\u6570\u636e\u6a21\u578b
\nclass PredictionRequest(BaseModel):
\n """\u9884\u6d4b\u8bf7\u6c42"""
\n model_name: str = Field(…, description="\u6a21\u578b\u540d\u79f0")
\n model_version: str = Field(…, description="\u6a21\u578b\u7248\u672c")
\n data: Union[Dict[str, Any], List[Any]] = Field(…, description="\u8f93\u5165\u6570\u636e")
\n request_id: Optional[str] = Field(None, description="\u8bf7\u6c42ID")
\n features: Optional[Dict[str, Any]] = Field(None, description="\u7279\u5f81\u6570\u636e")<\/p>\n

@validator('data')
\n def validate_data(cls, v):
\n if not v:
\n raise ValueError('\u6570\u636e\u4e0d\u80fd\u4e3a\u7a7a')
\n return v<\/p>\n

class BatchPredictionRequest(BaseModel):
\n """\u6279\u91cf\u9884\u6d4b\u8bf7\u6c42"""
\n model_name: str
\n model_version: str
\n data: List[Union[Dict[str, Any], List[Any]]]
\n request_id: Optional[str] = None<\/p>\n

class ModelLoadRequest(BaseModel):
\n """\u6a21\u578b\u52a0\u8f7d\u8bf7\u6c42"""
\n model_name: str
\n model_version: str
\n model_path: str
\n framework: str = "sklearn"<\/p>\n

class ABTestRequest(BaseModel):
\n """A\/B\u6d4b\u8bd5\u8bf7\u6c42"""
\n user_id: str
\n experiment_name: str
\n features: Dict[str, Any]
\n context: Optional[Dict[str, Any]] = None<\/p>\n

class PredictionResponse(BaseModel):
\n """\u9884\u6d4b\u54cd\u5e94"""
\n request_id: Optional[str]
\n prediction: Any
\n probabilities: Optional[List[float]] = None
\n model_name: str
\n model_version: str
\n latency_ms: float
\n timestamp: str
\n experiment_group: Optional[str] = None # A\/B\u6d4b\u8bd5\u5206\u7ec4<\/p>\n

class HealthResponse(BaseModel):
\n """\u5065\u5eb7\u68c0\u67e5\u54cd\u5e94"""
\n status: str
\n timestamp: str
\n uptime_seconds: float
\n model_count: int
\n healthy_model_count: int<\/p>\n

# \u521b\u5efaFastAPI\u5e94\u7528
\napp = FastAPI(
\n title="ML Model Serving API",
\n description="\u673a\u5668\u5b66\u4e60\u6a21\u578b\u5728\u7ebf\u670d\u52a1API",
\n version="1.0.0",
\n docs_url="\/docs",
\n redoc_url="\/redoc"
\n)<\/p>\n

# \u6dfb\u52a0CORS\u4e2d\u95f4\u4ef6
\napp.add_middleware(
\n CORSMiddleware,
\n allow_origins=["*"],
\n allow_credentials=True,
\n allow_methods=["*"],
\n allow_headers=["*"],
\n)<\/p>\n

# \u5168\u5c40\u53d8\u91cf
\nmodel_manager: Optional[ModelManager] = None
\nfeature_service: Optional[FeatureEngineeringService] = None
\nab_test_manager: Optional['ABTestManager'] = None # \u5c06\u5728\u540e\u9762\u5b9a\u4e49<\/p>\n

# \u4f9d\u8d56\u6ce8\u5165
\nasync def get_model_manager():
\n """\u83b7\u53d6\u6a21\u578b\u7ba1\u7406\u5668"""
\n return model_manager<\/p>\n

async def get_feature_service():
\n """\u83b7\u53d6\u7279\u5f81\u670d\u52a1"""
\n return feature_service<\/p>\n

async def get_ab_test_manager():
\n """\u83b7\u53d6A\/B\u6d4b\u8bd5\u7ba1\u7406\u5668"""
\n return ab_test_manager<\/p>\n

# \u542f\u52a8\u4e8b\u4ef6
\n@app.on_event("startup")
\nasync def startup_event():
\n """\u5e94\u7528\u542f\u52a8\u4e8b\u4ef6"""
\n global model_manager, feature_service, ab_test_manager<\/p>\n

# \u52a0\u8f7d\u914d\u7f6e
\n with open('config\/model_service_config.json', 'r') as f:
\n config = json.load(f)<\/p>\n

# \u521d\u59cb\u5316\u670d\u52a1
\n model_manager = ModelManager(config)
\n feature_service = FeatureEngineeringService()<\/p>\n

# \u521d\u59cb\u5316A\/B\u6d4b\u8bd5\u7ba1\u7406\u5668
\n ab_test_manager = ABTestManager(config.get('ab_test', {}))<\/p>\n

# \u52a0\u8f7d\u9ed8\u8ba4\u6a21\u578b
\n for model_config in config.get('models', []):
\n await model_manager.load_model(
\n model_config['name'],
\n model_config['version']
\n )<\/p>\n

logger.info("ML Model Serving API\u542f\u52a8\u5b8c\u6210")<\/p>\n

@app.on_event("shutdown")
\nasync def shutdown_event():
\n """\u5e94\u7528\u5173\u95ed\u4e8b\u4ef6"""
\n logger.info("\u6b63\u5728\u5173\u95edML Model Serving API…")<\/p>\n

# \u6e05\u7406\u8d44\u6e90
\n if model_manager:
\n model_manager.executor.shutdown(wait=True)<\/p>\n

# API\u8def\u7531
\n@app.get("\/")
\nasync def root():
\n """\u6839\u8def\u7531"""
\n return {
\n "service": "ML Model Serving API",
\n "version": "1.0.0",
\n "status": "running",
\n "timestamp": datetime.now().isoformat()
\n }<\/p>\n

@app.get("\/health")
\nasync def health_check(
\n mm: ModelManager = Depends(get_model_manager)
\n) -> HealthResponse:
\n """\u5065\u5eb7\u68c0\u67e5"""
\n if not mm:
\n raise HTTPException(status_code=503, detail="ModelManager\u672a\u521d\u59cb\u5316")<\/p>\n

health_status = await mm.health_check_all()<\/p>\n

return HealthResponse(
\n status="healthy" if health_status['healthy_models'] > 0 else "unhealthy",
\n timestamp=datetime.now().isoformat(),
\n uptime_seconds=health_status.get('uptime', 0),
\n model_count=health_status['total_models'],
\n healthy_model_count=health_status['healthy_models']
\n )<\/p>\n

@app.get("\/models")
\nasync def list_models(
\n mm: ModelManager = Depends(get_model_manager)
\n):
\n """\u5217\u51fa\u6240\u6709\u5df2\u52a0\u8f7d\u7684\u6a21\u578b"""
\n if not mm:
\n raise HTTPException(status_code=503, detail="ModelManager\u672a\u521d\u59cb\u5316")<\/p>\n

return {
\n "timestamp": datetime.now().isoformat(),
\n "models": mm.list_models()
\n }<\/p>\n

@app.post("\/models\/load")
\nasync def load_model(
\n request: ModelLoadRequest,
\n mm: ModelManager = Depends(get_model_manager),
\n background_tasks: BackgroundTasks = None
\n):
\n """\u52a0\u8f7d\u6a21\u578b"""
\n if not mm:
\n raise HTTPException(status_code=503, detail="ModelManager\u672a\u521d\u59cb\u5316")<\/p>\n

# \u5f02\u6b65\u52a0\u8f7d\u6a21\u578b
\n if background_tasks:
\n background_tasks.add_task(
\n mm.load_model,
\n request.model_name,
\n request.model_version
\n )
\n return {"message": "\u6a21\u578b\u52a0\u8f7d\u4efb\u52a1\u5df2\u63d0\u4ea4", "request_id": request.request_id}
\n else:
\n success = await mm.load_model(request.model_name, request.model_version)
\n if success:
\n return {"message": "\u6a21\u578b\u52a0\u8f7d\u6210\u529f", "request_id": request.request_id}
\n else:
\n raise HTTPException(status_code=500, detail="\u6a21\u578b\u52a0\u8f7d\u5931\u8d25")<\/p>\n

@app.post("\/predict")
\nasync def predict(
\n request: PredictionRequest,
\n mm: ModelManager = Depends(get_model_manager),
\n fs: FeatureEngineeringService = Depends(get_feature_service),
\n atm: 'ABTestManager' = Depends(get_ab_test_manager),
\n x_request_id: Optional[str] = Header(None, alias="X-Request-ID")
\n):
\n """\u5355\u6761\u9884\u6d4b"""
\n start_time = time.time()
\n request_id = request.request_id or x_request_id or str(uuid.uuid4())<\/p>\n

try:
\n # 1. \u7279\u5f81\u5de5\u7a0b（\u5982\u679c\u6709\u7684\u8bdd）
\n features = request.features
\n if not features and fs and request.data:
\n # \u5982\u679c\u63d0\u4f9b\u4e86\u539f\u59cb\u6570\u636e\u4f46\u6ca1\u6709\u7279\u5f81，\u5c1d\u8bd5\u7279\u5f81\u8f6c\u6362
\n # \u8fd9\u91cc\u9700\u8981\u6839\u636e\u5b9e\u9645\u60c5\u51b5\u5b9e\u73b0
\n pass<\/p>\n

# 2. \u6267\u884c\u9884\u6d4b
\n prediction_result = await mm.predict(
\n request.model_name,
\n request.model_version,
\n features or request.data
\n )<\/p>\n

# 3. \u5982\u679c\u662fA\/B\u6d4b\u8bd5，\u8bb0\u5f55\u5206\u7ec4\u4fe1\u606f
\n experiment_group = None
\n if atm and request_id:
\n experiment_group = atm.get_experiment_group(request_id)<\/p>\n

# 4. \u8bb0\u5f55\u9884\u6d4b\u65e5\u5fd7
\n await log_prediction(
\n request_id=request_id,
\n model_name=request.model_name,
\n model_version=request.model_version,
\n features=features,
\n prediction=prediction_result,
\n latency=time.time() – start_time
\n )<\/p>\n

# 5. \u8fd4\u56de\u7ed3\u679c
\n return PredictionResponse(
\n request_id=request_id,
\n prediction=prediction_result.get("prediction"),
\n probabilities=prediction_result.get("probabilities"),
\n model_name=prediction_result.get("model_name"),
\n model_version=prediction_result.get("model_version"),
\n latency_ms=prediction_result.get("latency_ms", 0),
\n timestamp=prediction_result.get("timestamp"),
\n experiment_group=experiment_group
\n )<\/p>\n

except Exception as e:
\n logger.error(f"\u9884\u6d4b\u5931\u8d25: {request_id}, \u9519\u8bef: {e}")
\n raise HTTPException(status_code=500, detail=str(e))<\/p>\n

@app.post("\/batch_predict")
\nasync def batch_predict(
\n request: BatchPredictionRequest,
\n mm: ModelManager = Depends(get_model_manager)
\n):
\n """\u6279\u91cf\u9884\u6d4b"""
\n if not mm:
\n raise HTTPException(status_code=503, detail="ModelManager\u672a\u521d\u59cb\u5316")<\/p>\n

try:
\n results = await mm.batch_predict(
\n request.model_name,
\n request.model_version,
\n request.data
\n )<\/p>\n

return {
\n "request_id": request.request_id,
\n "timestamp": datetime.now().isoformat(),
\n "total": len(results),
\n "success": sum(1 for r in results if "error" not in r),
\n "failed": sum(1 for r in results if "error" in r),
\n "results": results
\n }<\/p>\n

except Exception as e:
\n logger.error(f"\u6279\u91cf\u9884\u6d4b\u5931\u8d25: {e}")
\n raise HTTPException(status_code=500, detail=str(e))<\/p>\n

@app.post("\/ab_test\/predict")
\nasync def ab_test_predict(
\n request: ABTestRequest,
\n mm: ModelManager = Depends(get_model_manager),
\n atm: 'ABTestManager' = Depends(get_ab_test_manager)
\n):
\n """A\/B\u6d4b\u8bd5\u9884\u6d4b"""
\n if not mm or not atm:
\n raise HTTPException(status_code=503, detail="\u670d\u52a1\u672a\u521d\u59cb\u5316")<\/p>\n

request_id = str(uuid.uuid4())<\/p>\n

try:
\n # 1. \u83b7\u53d6A\/B\u6d4b\u8bd5\u5206\u914d
\n assignment = atm.assign_experiment(
\n user_id=request.user_id,
\n experiment_name=request.experiment_name,
\n context=request.context
\n )<\/p>\n

if not assignment:
\n raise HTTPException(status_code=400, detail="A\/B\u6d4b\u8bd5\u5206\u914d\u5931\u8d25")<\/p>\n

# 2. \u6839\u636e\u5206\u7ec4\u9009\u62e9\u6a21\u578b
\n model_name = assignment.get('model_name')
\n model_version = assignment.get('model_version')<\/p>\n

if not model_name or not model_version:
\n raise HTTPException(status_code=400, detail="A\/B\u6d4b\u8bd5\u914d\u7f6e\u9519\u8bef")<\/p>\n

# 3. \u6267\u884c\u9884\u6d4b
\n prediction_result = await mm.predict(
\n model_name,
\n model_version,
\n request.features
\n )<\/p>\n

# 4. \u8bb0\u5f55A\/B\u6d4b\u8bd5\u4e8b\u4ef6
\n atm.log_event(
\n request_id=request_id,
\n user_id=request.user_id,
\n experiment_name=request.experiment_name,
\n group_name=assignment.get('group_name'),
\n action='prediction',
\n metadata={
\n 'features': request.features,
\n 'prediction': prediction_result,
\n 'context': request.context
\n }
\n )<\/p>\n

# 5. \u8fd4\u56de\u7ed3\u679c
\n return PredictionResponse(
\n request_id=request_id,
\n prediction=prediction_result.get("prediction"),
\n probabilities=prediction_result.get("probabilities"),
\n model_name=prediction_result.get("model_name"),
\n model_version=prediction_result.get("model_version"),
\n latency_ms=prediction_result.get("latency_ms", 0),
\n timestamp=prediction_result.get("timestamp"),
\n experiment_group=assignment.get('group_name')
\n )<\/p>\n

except Exception as e:
\n logger.error(f"A\/B\u6d4b\u8bd5\u9884\u6d4b\u5931\u8d25: {e}")
\n raise HTTPException(status_code=500, detail=str(e))<\/p>\n

@app.get("\/ab_test\/experiments")
\nasync def list_experiments(
\n atm: 'ABTestManager' = Depends(get_ab_test_manager)
\n):
\n """\u5217\u51fa\u6240\u6709A\/B\u6d4b\u8bd5\u5b9e\u9a8c"""
\n if not atm:
\n raise HTTPException(status_code=503, detail="ABTestManager\u672a\u521d\u59cb\u5316")<\/p>\n

experiments = atm.list_experiments()<\/p>\n

return {
\n "timestamp": datetime.now().isoformat(),
\n "total": len(experiments),
\n "experiments": experiments
\n }<\/p>\n

@app.get("\/ab_test\/experiment\/{experiment_name}\/stats")
\nasync def get_experiment_stats(
\n experiment_name: str,
\n atm: 'ABTestManager' = Depends(get_ab_test_manager)
\n):
\n """\u83b7\u53d6\u5b9e\u9a8c\u7edf\u8ba1\u4fe1\u606f"""
\n if not atm:
\n raise HTTPException(status_code=503, detail="ABTestManager\u672a\u521d\u59cb\u5316")<\/p>\n

stats = atm.get_experiment_stats(experiment_name)<\/p>\n

if not stats:
\n raise HTTPException(status_code=404, detail="\u5b9e\u9a8c\u4e0d\u5b58\u5728")<\/p>\n

return stats<\/p>\n

async def log_prediction(
\n request_id: str,
\n model_name: str,
\n model_version: str,
\n features: Any,
\n prediction: Any,
\n latency: float
\n):
\n """\u8bb0\u5f55\u9884\u6d4b\u65e5\u5fd7"""
\n # \u8fd9\u91cc\u53ef\u4ee5\u5b9e\u73b0\u65e5\u5fd7\u8bb0\u5f55\u903b\u8f91
\n # \u53ef\u4ee5\u8bb0\u5f55\u5230\u6587\u4ef6\u3001\u6570\u636e\u5e93\u6216\u76d1\u63a7\u7cfb\u7edf
\n log_entry = {
\n "request_id": request_id,
\n "timestamp": datetime.now().isoformat(),
\n "model": f"{model_name}:{model_version}",
\n "features": features,
\n "prediction": prediction,
\n "latency": latency,
\n "type": "prediction"
\n }<\/p>\n

# \u5f02\u6b65\u5199\u5165\u65e5\u5fd7
\n asyncio.create_task(_write_log(log_entry))<\/p>\n

async def _write_log(log_entry: Dict[str, Any]):
\n """\u5199\u5165\u65e5\u5fd7（\u5f02\u6b65）"""
\n try:
\n # \u8fd9\u91cc\u53ef\u4ee5\u5b9e\u73b0\u5b9e\u9645\u7684\u65e5\u5fd7\u5199\u5165\u903b\u8f91
\n # \u4f8b\u5982\u5199\u5165\u6587\u4ef6\u3001\u6570\u636e\u5e93\u6216\u53d1\u9001\u5230\u65e5\u5fd7\u670d\u52a1
\n logger.info(f"\u9884\u6d4b\u65e5\u5fd7: {json.dumps(log_entry)}")
\n except Exception as e:
\n logger.error(f"\u65e5\u5fd7\u5199\u5165\u5931\u8d25: {e}")<\/p>\n

# \u4e2d\u95f4\u4ef6：\u8bf7\u6c42ID\u6ce8\u5165
\n@app.middleware("http")
\nasync def add_request_id(request, call_next):
\n """\u6dfb\u52a0\u8bf7\u6c42ID\u4e2d\u95f4\u4ef6"""
\n request_id = request.headers.get('X-Request-ID') or str(uuid.uuid4())<\/p>\n

# \u5c06\u8bf7\u6c42ID\u6dfb\u52a0\u5230\u8bf7\u6c42\u72b6\u6001
\n request.state.request_id = request_id<\/p>\n

# \u5904\u7406\u8bf7\u6c42
\n start_time = time.time()
\n response = await call_next(request)
\n process_time = time.time() – start_time<\/p>\n

# \u6dfb\u52a0\u8bf7\u6c42ID\u5230\u54cd\u5e94\u5934
\n response.headers['X-Request-ID'] = request_id
\n response.headers['X-Process-Time'] = str(process_time)<\/p>\n

return response<\/p>\n

# \u9519\u8bef\u5904\u7406
\n@app.exception_handler(Exception)
\nasync def global_exception_handler(request, exc):
\n """\u5168\u5c40\u5f02\u5e38\u5904\u7406\u5668"""
\n request_id = getattr(request.state, 'request_id', 'unknown')<\/p>\n

logger.error(f"\u8bf7\u6c42\u5f02\u5e38: {request_id}, \u9519\u8bef: {exc}")<\/p>\n

return JSONResponse(
\n status_code=500,
\n content={
\n "request_id": request_id,
\n "error": str(exc),
\n "timestamp": datetime.now().isoformat()
\n }
\n )<\/p>\n

def start_server(host: str = "0.0.0.0", port: int = 8000):
\n """\u542f\u52a8\u670d\u52a1\u5668"""
\n uvicorn.run(
\n app,
\n host=host,
\n port=port,
\n log_level="info",
\n access_log=True
\n )<\/p>\n

if __name__ == "__main__":
\n import argparse<\/p>\n

parser = argparse.ArgumentParser(description="ML Model Serving API")
\n parser.add_argument("–host", default="0.0.0.0", help="\u670d\u52a1\u5668\u5730\u5740")
\n parser.add_argument("–port", type=int, default=8000, help="\u670d\u52a1\u5668\u7aef\u53e3")<\/p>\n

args = parser.parse_args()
\n start_server(args.host, args.port) <\/p>\n

\u4e8c\u3001A\/B\u6d4b\u8bd5\u7cfb\u7edf\u5b9e\u73b0<\/h3>\n

2.1 A\/B\u6d4b\u8bd5\u6838\u5fc3\u67b6\u6784<\/h4>\n

python<\/p>\n<\/p>\n

\u590d\u5236<\/p>\n<\/p>\n

\u4e0b\u8f7d<\/p>\n

"""
\nA\/B\u6d4b\u8bd5\u7cfb\u7edf\u6838\u5fc3\u5b9e\u73b0
\n"""
\nimport hashlib
\nimport random
\nimport json
\nfrom typing import Dict, List, Optional, Any, Union
\nfrom dataclasses import dataclass, field
\nfrom datetime import datetime, timedelta
\nfrom enum import Enum
\nimport asyncio
\nfrom collections import defaultdict
\nimport statistics
\nimport numpy as np
\nfrom scipy import stats # \u7528\u4e8e\u7edf\u8ba1\u68c0\u9a8c<\/p>\n

# ==================== \u6570\u636e\u6a21\u578b ====================
\nclass ExperimentStatus(Enum):
\n """\u5b9e\u9a8c\u72b6\u6001"""
\n DRAFT = "draft" # \u8349\u7a3f
\n RUNNING = "running" # \u8fd0\u884c\u4e2d
\n PAUSED = "paused" # \u6682\u505c
\n STOPPED = "stopped" # \u505c\u6b62
\n COMPLETED = "completed" # \u5b8c\u6210<\/p>\n

class AssignmentAlgorithm(Enum):
\n """\u5206\u914d\u7b97\u6cd5"""
\n RANDOM = "random" # \u968f\u673a\u5206\u914d
\n HASH_BASED = "hash_based" # \u57fa\u4e8e\u54c8\u5e0c\u7684\u5206\u914d
\n WEIGHTED = "weighted" # \u52a0\u6743\u5206\u914d
\n BANDIT = "bandit" # \u591a\u81c2\u8001\u864e\u673a\u7b97\u6cd5
\n CUSTOM = "custom" # \u81ea\u5b9a\u4e49\u7b97\u6cd5<\/p>\n

@dataclass
\nclass ExperimentGroup:
\n """\u5b9e\u9a8c\u5206\u7ec4"""
\n name: str # \u5206\u7ec4\u540d\u79f0
\n weight: float = 1.0 # \u5206\u914d\u6743\u91cd
\n model_name: Optional[str] = None # \u4f7f\u7528\u7684\u6a21\u578b\u540d\u79f0
\n model_version: Optional[str] = None # \u4f7f\u7528\u7684\u6a21\u578b\u7248\u672c
\n parameters: Dict[str, Any] = field(default_factory=dict) # \u81ea\u5b9a\u4e49\u53c2\u6570
\n description: Optional[str] = None<\/p>\n

@property
\n def model_key(self) -> Optional[str]:
\n """\u6a21\u578b\u952e\u503c"""
\n if self.model_name and self.model_version:
\n return f"{self.model_name}:{self.model_version}"
\n return None<\/p>\n

@dataclass
\nclass Experiment:
\n """A\/B\u6d4b\u8bd5\u5b9e\u9a8c"""
\n name: str # \u5b9e\u9a8c\u540d\u79f0
\n description: Optional[str] = None # \u5b9e\u9a8c\u63cf\u8ff0
\n status: ExperimentStatus = ExperimentStatus.DRAFT # \u5b9e\u9a8c\u72b6\u6001
\n start_time: Optional[datetime] = None # \u5f00\u59cb\u65f6\u95f4
\n end_time: Optional[datetime] = None # \u7ed3\u675f\u65f6\u95f4
\n groups: List[ExperimentGroup] = field(default_factory=list) # \u5b9e\u9a8c\u5206\u7ec4
\n assignment_algorithm: AssignmentAlgorithm = AssignmentAlgorithm.RANDOM # \u5206\u914d\u7b97\u6cd5
\n target_users: Optional[List[str]] = None # \u76ee\u6807\u7528\u6237
\n sample_rate: float = 1.0 # \u91c7\u6837\u7387
\n metrics: List[str] = field(default_factory=list) # \u76d1\u63a7\u6307\u6807
\n hypotheses: Optional[str] = None # \u5b9e\u9a8c\u5047\u8bbe
\n created_at: datetime = field(default_factory=datetime.now)
\n updated_at: datetime = field(default_factory=datetime.now)<\/p>\n

def get_group(self, group_name: str) -> Optional[ExperimentGroup]:
\n """\u83b7\u53d6\u5206\u7ec4"""
\n for group in self.groups:
\n if group.name == group_name:
\n return group
\n return None<\/p>\n

def is_running(self) -> bool:
\n """\u662f\u5426\u6b63\u5728\u8fd0\u884c"""
\n if self.status != ExperimentStatus.RUNNING:
\n return False<\/p>\n

now = datetime.now()
\n if self.start_time and now < self.start_time:
\n return False
\n if self.end_time and now > self.end_time:
\n return False<\/p>\n

return True<\/p>\n

@dataclass
\nclass Assignment:
\n """\u7528\u6237\u5206\u914d"""
\n experiment_name: str # \u5b9e\u9a8c\u540d\u79f0
\n user_id: str # \u7528\u6237ID
\n group_name: str # \u5206\u7ec4\u540d\u79f0
\n assigned_at: datetime # \u5206\u914d\u65f6\u95f4
\n assignment_id: str # \u5206\u914dID
\n context: Optional[Dict[str, Any]] = None # \u5206\u914d\u4e0a\u4e0b\u6587<\/p>\n

@dataclass
\nclass Event:
\n """\u5b9e\u9a8c\u4e8b\u4ef6"""
\n event_id: str # \u4e8b\u4ef6ID
\n experiment_name: str # \u5b9e\u9a8c\u540d\u79f0
\n user_id: str # \u7528\u6237ID
\n group_name: str # \u5206\u7ec4\u540d\u79f0
\n event_type: str # \u4e8b\u4ef6\u7c7b\u578b
\n timestamp: datetime # \u4e8b\u4ef6\u65f6\u95f4
\n metadata: Dict[str, Any] = field(default_factory=dict) # \u4e8b\u4ef6\u5143\u6570\u636e<\/p>\n

@dataclass
\nclass ExperimentStats:
\n """\u5b9e\u9a8c\u7edf\u8ba1"""
\n experiment_name: str
\n start_time: datetime
\n end_time: Optional[datetime]
\n total_users: int
\n total_events: int
\n group_stats: Dict[str, 'GroupStats'] # \u5206\u7ec4\u7edf\u8ba1
\n significance_test: Optional[Dict[str, Any]] = None # \u663e\u8457\u6027\u68c0\u9a8c\u7ed3\u679c<\/p>\n

@dataclass
\nclass GroupStats:
\n """\u5206\u7ec4\u7edf\u8ba1"""
\n group_name: str
\n user_count: int
\n event_counts: Dict[str, int] # \u4e8b\u4ef6\u8ba1\u6570
\n metric_values: Dict[str, List[float]] # \u6307\u6807\u503c
\n conversions: Dict[str, float] # \u8f6c\u5316\u7387<\/p>\n

@property
\n def conversion_rate(self) -> float:
\n """\u603b\u4f53\u8f6c\u5316\u7387"""
\n if 'conversion' in self.event_counts:
\n total_events = sum(self.event_counts.values())
\n if total_events > 0:
\n return self.event_counts['conversion'] \/ total_events
\n return 0.0<\/p>\n

# ==================== \u5206\u914d\u7b97\u6cd5\u5b9e\u73b0 ====================
\nclass AssignmentAlgorithmBase(ABC):
\n """\u5206\u914d\u7b97\u6cd5\u57fa\u7c7b"""<\/p>\n

@abstractmethod
\n def assign(self, experiment: Experiment, user_id: str, context: Optional[Dict] = None) -> Optional[str]:
\n """\u5206\u914d\u7528\u6237\u5230\u5206\u7ec4"""
\n pass<\/p>\n

class RandomAssignment(AssignmentAlgorithmBase):
\n """\u968f\u673a\u5206\u914d\u7b97\u6cd5"""<\/p>\n

def assign(self, experiment: Experiment, user_id: str, context: Optional[Dict] = None) -> Optional[str]:
\n """\u968f\u673a\u5206\u914d"""
\n if not experiment.groups:
\n return None<\/p>\n

# \u8ba1\u7b97\u603b\u6743\u91cd
\n total_weight = sum(group.weight for group in experiment.groups)
\n if total_weight <= 0:
\n return None<\/p>\n

# \u968f\u673a\u9009\u62e9
\n rand = random.random() * total_weight
\n cumulative = 0<\/p>\n

for group in experiment.groups:
\n cumulative += group.weight
\n if rand <= cumulative:
\n return group.name<\/p>\n

# \u7406\u8bba\u4e0a\u4e0d\u4f1a\u6267\u884c\u5230\u8fd9\u91cc
\n return experiment.groups[-1].name<\/p>\n

class HashBasedAssignment(AssignmentAlgorithmBase):
\n """\u57fa\u4e8e\u54c8\u5e0c\u7684\u5206\u914d\u7b97\u6cd5（\u786e\u4fdd\u4e00\u81f4\u6027）"""<\/p>\n

def assign(self, experiment: Experiment, user_id: str, context: Optional[Dict] = None) -> Optional[str]:
\n """\u54c8\u5e0c\u5206\u914d"""
\n if not experiment.groups:
\n return None<\/p>\n

# \u4f7f\u7528\u7528\u6237ID\u548c\u5b9e\u9a8c\u540d\u79f0\u751f\u6210\u54c8\u5e0c
\n hash_input = f"{experiment.name}:{user_id}"
\n if context:
\n # \u53ef\u4ee5\u5305\u542b\u4e0a\u4e0b\u6587\u4fe1\u606f
\n hash_input += f":{json.dumps(context, sort_keys=True)}"<\/p>\n

# \u751f\u6210\u54c8\u5e0c\u503c
\n hash_val = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)<\/p>\n

# \u8ba1\u7b97\u603b\u6743\u91cd
\n total_weight = sum(group.weight for group in experiment.groups)
\n if total_weight <= 0:
\n return None<\/p>\n

# \u57fa\u4e8e\u54c8\u5e0c\u503c\u9009\u62e9\u5206\u7ec4
\n hash_mod = hash_val % total_weight
\n cumulative = 0<\/p>\n

for group in experiment.groups:
\n cumulative += group.weight
\n if hash_mod < cumulative:
\n return group.name<\/p>\n

return experiment.groups[-1].name<\/p>\n

class WeightedAssignment(AssignmentAlgorithmBase):
\n """\u52a0\u6743\u5206\u914d\u7b97\u6cd5"""<\/p>\n

def assign(self, experiment: Experiment, user_id: str, context: Optional[Dict] = None) -> Optional[str]:
\n """\u52a0\u6743\u5206\u914d"""
\n if not experiment.groups:
\n return None<\/p>\n

# \u83b7\u53d6\u5386\u53f2\u5206\u914d\u6570\u636e（\u5b9e\u9645\u5b9e\u73b0\u4e2d\u5e94\u4ece\u6570\u636e\u5e93\u83b7\u53d6）
\n historical_assignments = self._get_historical_assignments(experiment.name)<\/p>\n

# \u8ba1\u7b97\u6bcf\u4e2a\u5206\u7ec4\u7684\u5f53\u524d\u5206\u914d\u6bd4\u4f8b
\n group_counts = defaultdict(int)
\n for assignment in historical_assignments:
\n group_counts[assignment.group_name] += 1<\/p>\n

total_assignments = sum(group_counts.values())<\/p>\n

# \u8c03\u6574\u6743\u91cd\u4ee5\u5b9e\u73b0\u76ee\u6807\u5206\u914d\u6bd4\u4f8b
\n adjusted_weights = []
\n for group in experiment.groups:
\n expected_ratio = group.weight \/ sum(g.weight for g in experiment.groups)
\n actual_ratio = group_counts[group.name] \/ total_assignments if total_assignments > 0 else 0<\/p>\n

# \u5982\u679c\u5b9e\u9645\u6bd4\u4f8b\u4f4e\u4e8e\u9884\u671f，\u589e\u52a0\u5206\u914d\u6982\u7387
\n adjustment = max(0.1, expected_ratio – actual_ratio + 1.0)
\n adjusted_weights.append(adjustment)<\/p>\n

# \u57fa\u4e8e\u8c03\u6574\u540e\u7684\u6743\u91cd\u8fdb\u884c\u5206\u914d
\n total_adjusted = sum(adjusted_weights)
\n if total_adjusted <= 0:
\n return None<\/p>\n

rand = random.random() * total_adjusted
\n cumulative = 0<\/p>\n

for i, group in enumerate(experiment.groups):
\n cumulative += adjusted_weights[i]
\n if rand <= cumulative:
\n return group.name<\/p>\n

return experiment.groups[-1].name<\/p>\n

def _get_historical_assignments(self, experiment_name: str) -> List[Assignment]:
\n """\u83b7\u53d6\u5386\u53f2\u5206\u914d\u6570\u636e（\u7b80\u5316\u5b9e\u73b0）"""
\n # \u5b9e\u9645\u5b9e\u73b0\u4e2d\u5e94\u4ece\u6570\u636e\u5e93\u67e5\u8be2
\n return []<\/p>\n

class BanditAssignment(AssignmentAlgorithmBase):
\n """\u591a\u81c2\u8001\u864e\u673a\u5206\u914d\u7b97\u6cd5"""<\/p>\n

def __init__(self, alpha: float = 1.0, beta: float = 1.0):
\n self.alpha = alpha # \u6210\u529f\u5148\u9a8c\u53c2\u6570
\n self.beta = beta # \u5931\u8d25\u5148\u9a8c\u53c2\u6570
\n self.group_stats = defaultdict(lambda: {'success': 0, 'failure': 0})<\/p>\n

def assign(self, experiment: Experiment, user_id: str, context: Optional[Dict] = None) -> Optional[str]:
\n """Bandit\u5206\u914d"""
\n if not experiment.groups:
\n return None<\/p>\n

# \u5982\u679c\u67d0\u4e2a\u5206\u7ec4\u8fd8\u6ca1\u6709\u6570\u636e，\u4f18\u5148\u63a2\u7d22
\n unexplored_groups = []
\n for group in experiment.groups:
\n stats = self.group_stats[group.name]
\n if stats['success'] + stats['failure'] == 0:
\n unexplored_groups.append(group.name)<\/p>\n

if unexplored_groups:
\n # \u968f\u673a\u9009\u62e9\u4e00\u4e2a\u672a\u63a2\u7d22\u7684\u5206\u7ec4
\n return random.choice(unexplored_groups)<\/p>\n

# \u57fa\u4e8eBeta\u5206\u5e03\u91c7\u6837\u9009\u62e9\u5206\u7ec4
\n samples = []
\n for group in experiment.groups:
\n stats = self.group_stats[group.name]
\n # \u4eceBeta\u5206\u5e03\u91c7\u6837
\n sample = np.random.beta(
\n stats['success'] + self.alpha,
\n stats['failure'] + self.beta
\n )
\n samples.append((sample, group.name))<\/p>\n

# \u9009\u62e9\u91c7\u6837\u503c\u6700\u5927\u7684\u5206\u7ec4
\n samples.sort(reverse=True)
\n return samples[0][1]<\/p>\n

def update(self, group_name: str, success: bool):
\n """\u66f4\u65b0\u5206\u7ec4\u7edf\u8ba1"""
\n if success:
\n self.group_stats[group_name]['success'] += 1
\n else:
\n self.group_stats[group_name]['failure'] += 1<\/p>\n

# ==================== A\/B\u6d4b\u8bd5\u7ba1\u7406\u5668 ====================
\nclass ABTestManager:
\n """A\/B\u6d4b\u8bd5\u7ba1\u7406\u5668"""<\/p>\n

def __init__(self, config: Dict[str, Any]):
\n self.config = config
\n self.experiments: Dict[str, Experiment] = {}
\n self.assignments: Dict[str, Assignment] = {} # assignment_id -> Assignment
\n self.events: List[Event] = []<\/p>\n

# \u5206\u914d\u7b97\u6cd5\u6ce8\u518c\u8868
\n self.algorithms = {
\n AssignmentAlgorithm.RANDOM: RandomAssignment(),
\n AssignmentAlgorithm.HASH_BASED: HashBasedAssignment(),
\n AssignmentAlgorithm.WEIGHTED: WeightedAssignment(),
\n AssignmentAlgorithm.BANDIT: BanditAssignment()
\n }<\/p>\n

self.logger = logging.getLogger("ABTestManager")<\/p>\n

# \u52a0\u8f7d\u5b9e\u9a8c\u914d\u7f6e
\n self._load_experiments()<\/p>\n

def _load_experiments(self):
\n """\u52a0\u8f7d\u5b9e\u9a8c\u914d\u7f6e"""
\n experiments_config = self.config.get('experiments', [])<\/p>\n

for exp_config in experiments_config:
\n experiment = Experiment(
\n name=exp_config['name'],
\n description=exp_config.get('description'),
\n status=ExperimentStatus(exp_config.get('status', 'draft')),
\n groups=[
\n ExperimentGroup(
\n name=g['name'],
\n weight=g.get('weight', 1.0),
\n model_name=g.get('model_name'),
\n model_version=g.get('model_version'),
\n parameters=g.get('parameters', {}),
\n description=g.get('description')
\n )
\n for g in exp_config.get('groups', [])
\n ],
\n assignment_algorithm=AssignmentAlgorithm(
\n exp_config.get('assignment_algorithm', 'random')
\n ),
\n target_users=exp_config.get('target_users'),
\n sample_rate=exp_config.get('sample_rate', 1.0),
\n metrics=exp_config.get('metrics', []),
\n hypotheses=exp_config.get('hypotheses')
\n )<\/p>\n

self.experiments[experiment.name] = experiment
\n self.logger.info(f"\u52a0\u8f7d\u5b9e\u9a8c: {experiment.name}")<\/p>\n

def create_experiment(self, experiment: Experiment) -> bool:
\n """\u521b\u5efa\u5b9e\u9a8c"""
\n if experiment.name in self.experiments:
\n self.logger.error(f"\u5b9e\u9a8c\u5df2\u5b58\u5728: {experiment.name}")
\n return False<\/p>\n

# \u9a8c\u8bc1\u5b9e\u9a8c\u914d\u7f6e
\n if not self._validate_experiment(experiment):
\n return False<\/p>\n

self.experiments[experiment.name] = experiment
\n self.logger.info(f"\u521b\u5efa\u5b9e\u9a8c: {experiment.name}")<\/p>\n

# \u4fdd\u5b58\u5230\u914d\u7f6e\u6587\u4ef6\u6216\u6570\u636e\u5e93
\n self._save_experiments()<\/p>\n

return True<\/p>\n

def start_experiment(self, experiment_name: str) -> bool:
\n """\u542f\u52a8\u5b9e\u9a8c"""
\n if experiment_name not in self.experiments:
\n self.logger.error(f"\u5b9e\u9a8c\u4e0d\u5b58\u5728: {experiment_name}")
\n return False<\/p>\n

experiment = self.experiments[experiment_name]<\/p>\n

if experiment.status == ExperimentStatus.RUNNING:
\n self.logger.warning(f"\u5b9e\u9a8c\u5df2\u5728\u8fd0\u884c: {experiment_name}")
\n return True<\/p>\n

experiment.status = ExperimentStatus.RUNNING
\n experiment.start_time = datetime.now()
\n experiment.updated_at = datetime.now()<\/p>\n

self.logger.info(f"\u542f\u52a8\u5b9e\u9a8c: {experiment_name}")
\n self._save_experiments()<\/p>\n

return True<\/p>\n

def stop_experiment(self, experiment_name: str) -> bool:
\n """\u505c\u6b62\u5b9e\u9a8c"""
\n if experiment_name not in self.experiments:
\n self.logger.error(f"\u5b9e\u9a8c\u4e0d\u5b58\u5728: {experiment_name}")
\n return False<\/p>\n

experiment = self.experiments[experiment_name]<\/p>\n

if experiment.status == ExperimentStatus.STOPPED:
\n return True<\/p>\n

experiment.status = ExperimentStatus.STOPPED
\n experiment.end_time = datetime.now()
\n experiment.updated_at = datetime.now()<\/p>\n

self.logger.info(f"\u505c\u6b62\u5b9e\u9a8c: {experiment_name}")
\n self._save_experiments()<\/p>\n

return True<\/p>\n

def assign_experiment(
\n self,
\n user_id: str,
\n experiment_name: str,
\n context: Optional[Dict[str, Any]] = None
\n ) -> Optional[Dict[str, Any]]:
\n """\u5206\u914d\u7528\u6237\u5230\u5b9e\u9a8c\u5206\u7ec4"""
\n if experiment_name not in self.experiments:
\n self.logger.error(f"\u5b9e\u9a8c\u4e0d\u5b58\u5728: {experiment_name}")
\n return None<\/p>\n

experiment = self.experiments[experiment_name]<\/p>\n

# \u68c0\u67e5\u5b9e\u9a8c\u72b6\u6001
\n if not experiment.is_running():
\n self.logger.warning(f"\u5b9e\u9a8c\u672a\u8fd0\u884c: {experiment_name}")
\n return None<\/p>\n

# \u68c0\u67e5\u76ee\u6807\u7528\u6237
\n if experiment.target_users and user_id not in experiment.target_users:
\n self.logger.debug(f"\u7528\u6237\u4e0d\u5728\u76ee\u6807\u5217\u8868\u4e2d: {user_id}")
\n return None<\/p>\n

# \u68c0\u67e5\u91c7\u6837\u7387
\n if experiment.sample_rate < 1.0:
\n hash_input = f"{experiment_name}:{user_id}"
\n hash_val = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
\n if (hash_val % 1000) \/ 1000.0 > experiment.sample_rate:
\n self.logger.debug(f"\u7528\u6237\u672a\u91c7\u6837: {user_id}")
\n return None<\/p>\n

# \u83b7\u53d6\u5206\u914d\u7b97\u6cd5
\n algorithm = self.algorithms.get(experiment.assignment_algorithm)
\n if not algorithm:
\n self.logger.error(f"\u4e0d\u652f\u6301\u7684\u5206\u914d\u7b97\u6cd5: {experiment.assignment_algorithm}")
\n algorithm = RandomAssignment()<\/p>\n

# \u5206\u914d\u5206\u7ec4
\n group_name = algorithm.assign(experiment, user_id, context)
\n if not group_name:
\n self.logger.error(f"\u5206\u914d\u5931\u8d25: {experiment_name}, {user_id}")
\n return None<\/p>\n

group = experiment.get_group(group_name)
\n if not group:
\n self.logger.error(f"\u5206\u7ec4\u4e0d\u5b58\u5728: {group_name}")
\n return None<\/p>\n

# \u521b\u5efa\u5206\u914d\u8bb0\u5f55
\n assignment_id = hashlib.md5(
\n f"{experiment_name}:{user_id}:{datetime.now().isoformat()}".encode()
\n ).hexdigest()<\/p>\n

assignment = Assignment(
\n experiment_name=experiment_name,
\n user_id=user_id,
\n group_name=group_name,
\n assigned_at=datetime.now(),
\n assignment_id=assignment_id,
\n context=context
\n )<\/p>\n

self.assignments[assignment_id] = assignment<\/p>\n

# \u8bb0\u5f55\u5206\u914d\u4e8b\u4ef6
\n self.log_event(
\n request_id=assignment_id,
\n user_id=user_id,
\n experiment_name=experiment_name,
\n group_name=group_name,
\n action='assignment',
\n metadata={
\n 'algorithm': experiment.assignment_algorithm.value,
\n 'context': context
\n }
\n )<\/p>\n

self.logger.debug(f"\u5206\u914d\u7528\u6237 {user_id} \u5230\u5b9e\u9a8c {experiment_name} \u5206\u7ec4 {group_name}")<\/p>\n

return {
\n 'experiment_name': experiment_name,
\n 'group_name': group_name,
\n 'assignment_id': assignment_id,
\n 'model_name': group.model_name,
\n 'model_version': group.model_version,
\n 'parameters': group.parameters
\n }<\/p>\n

def get_experiment_group(self, request_id: str) -> Optional[str]:
\n """\u83b7\u53d6\u5b9e\u9a8c\u5206\u7ec4（\u901a\u8fc7\u8bf7\u6c42ID）"""
\n # \u5728\u5b9e\u9645\u5b9e\u73b0\u4e2d，\u9700\u8981\u5efa\u7acbrequest_id\u5230assignment\u7684\u6620\u5c04
\n # \u8fd9\u91cc\u7b80\u5316\u5b9e\u73b0，\u5047\u8bberequest_id\u5c31\u662fassignment_id
\n if request_id in self.assignments:
\n assignment = self.assignments[request_id]
\n return assignment.group_name
\n return None<\/p>\n

def log_event(
\n self,
\n request_id: str,
\n user_id: str,
\n experiment_name: str,
\n group_name: str,
\n action: str,
\n metadata: Optional[Dict[str, Any]] = None
\n ):
\n """\u8bb0\u5f55\u5b9e\u9a8c\u4e8b\u4ef6"""
\n event = Event(
\n event_id=hashlib.md5(
\n f"{request_id}:{action}:{datetime.now().isoformat()}".encode()
\n ).hexdigest(),
\n experiment_name=experiment_name,
\n user_id=user_id,
\n group_name=group_name,
\n event_type=action,
\n timestamp=datetime.now(),
\n metadata=metadata or {}
\n )<\/p>\n

self.events.append(event)<\/p>\n

# \u5982\u679c\u662fBandit\u7b97\u6cd5，\u66f4\u65b0\u7edf\u8ba1
\n if experiment_name in self.experiments:
\n experiment = self.experiments[experiment_name]
\n if experiment.assignment_algorithm == AssignmentAlgorithm.BANDIT:
\n algorithm = self.algorithms.get(AssignmentAlgorithm.BANDIT)
\n if algorithm and isinstance(algorithm, BanditAssignment):
\n # \u5224\u65ad\u662f\u5426\u6210\u529f（\u6839\u636e\u4e1a\u52a1\u903b\u8f91）
\n success = self._is_success_event(action, metadata)
\n algorithm.update(group_name, success)<\/p>\n

self.logger.debug(f"\u8bb0\u5f55\u4e8b\u4ef6: {experiment_name}, {group_name}, {action}")<\/p>\n

def _is_success_event(self, action: str, metadata: Optional[Dict]) -> bool:
\n """\u5224\u65ad\u4e8b\u4ef6\u662f\u5426\u6210\u529f"""
\n # \u8fd9\u91cc\u9700\u8981\u6839\u636e\u4e1a\u52a1\u903b\u8f91\u5b9e\u73b0
\n # \u4f8b\u5982：\u5982\u679c\u662f\u8d2d\u4e70\u4e8b\u4ef6，\u4e14\u91d1\u989d\u5927\u4e8e0，\u5219\u8ba4\u4e3a\u662f\u6210\u529f
\n if action == 'purchase' and metadata:
\n amount = metadata.get('amount', 0)
\n return amount > 0
\n elif action == 'conversion':
\n return True
\n return False<\/p>\n

def list_experiments(self) -> List[Dict[str, Any]]:
\n """\u5217\u51fa\u6240\u6709\u5b9e\u9a8c"""
\n result = []
\n for experiment in self.experiments.values():
\n result.append({
\n 'name': experiment.name,
\n 'description': experiment.description,
\n 'status': experiment.status.value,
\n 'start_time': experiment.start_time.isoformat() if experiment.start_time else None,
\n 'end_time': experiment.end_time.isoformat() if experiment.end_time else None,
\n 'groups': [
\n {
\n 'name': g.name,
\n 'weight': g.weight,
\n 'model_name': g.model_name,
\n 'model_version': g.model_version
\n }
\n for g in experiment.groups
\n ],
\n 'assignment_algorithm': experiment.assignment_algorithm.value,
\n 'sample_rate': experiment.sample_rate,
\n 'created_at': experiment.created_at.isoformat(),
\n 'updated_at': experiment.updated_at.isoformat()
\n })
\n return result<\/p>\n

def get_experiment_stats(self, experiment_name: str) -> Optional[Dict[str, Any]]:
\n """\u83b7\u53d6\u5b9e\u9a8c\u7edf\u8ba1\u4fe1\u606f"""
\n if experiment_name not in self.experiments:
\n return None<\/p>\n

experiment = self.experiments[experiment_name]<\/p>\n

# \u6536\u96c6\u5206\u914d\u6570\u636e
\n experiment_assignments = [
\n a for a in self.assignments.values()
\n if a.experiment_name == experiment_name
\n ]<\/p>\n

# \u6536\u96c6\u4e8b\u4ef6\u6570\u636e
\n experiment_events = [
\n e for e in self.events
\n if e.experiment_name == experiment_name
\n ]<\/p>\n

# \u6309\u5206\u7ec4\u7edf\u8ba1
\n group_stats = {}
\n for group in experiment.groups:
\n group_assignments = [
\n a for a in experiment_assignments
\n if a.group_name == group.name
\n ]<\/p>\n

group_events = [
\n e for e in experiment_events
\n if e.group_name == group.name
\n ]<\/p>\n

# \u8ba1\u7b97\u4e8b\u4ef6\u8ba1\u6570
\n event_counts = defaultdict(int)
\n metric_values = defaultdict(list)<\/p>\n

for event in group_events:
\n event_counts[event.event_type] += 1<\/p>\n

# \u63d0\u53d6\u6307\u6807\u503c
\n for metric in experiment.metrics:
\n if metric in event.metadata:
\n value = event.metadata[metric]
\n if isinstance(value, (int, float)):
\n metric_values[metric].append(value)<\/p>\n

# \u8ba1\u7b97\u8f6c\u5316\u7387
\n conversions = {}
\n total_users = len(set(a.user_id for a in group_assignments))
\n if total_users > 0:
\n for event_type in event_counts:
\n # \u8ba1\u7b97\u6bcf\u4e2a\u7528\u6237\u7684\u5e73\u5747\u4e8b\u4ef6\u6570
\n conversions[event_type] = event_counts[event_type] \/ total_users<\/p>\n

group_stats[group.name] = GroupStats(
\n group_name=group.name,
\n user_count=total_users,
\n event_counts=dict(event_counts),
\n metric_values=dict(metric_values),
\n conversions=conversions
\n )<\/p>\n

# \u663e\u8457\u6027\u68c0\u9a8c
\n significance_test = None
\n if len(experiment.groups) >= 2:
\n significance_test = self._calculate_significance(
\n experiment, group_stats
\n )<\/p>\n

stats = ExperimentStats(
\n experiment_name=experiment_name,
\n start_time=experiment.start_time or experiment.created_at,
\n end_time=experiment.end_time,
\n total_users=len(set(a.user_id for a in experiment_assignments)),
\n total_events=len(experiment_events),
\n group_stats=group_stats,
\n significance_test=significance_test
\n )<\/p>\n

return self._format_stats(stats)<\/p>\n

def _calculate_significance(
\n self,
\n experiment: Experiment,
\n group_stats: Dict[str, GroupStats]
\n ) -> Dict[str, Any]:
\n """\u8ba1\u7b97\u663e\u8457\u6027\u68c0\u9a8c"""
\n if len(experiment.groups) < 2:
\n return None<\/p>\n

# \u9009\u62e9\u63a7\u5236\u7ec4（\u901a\u5e38\u7b2c\u4e00\u4e2a\u5206\u7ec4）
\n control_group = experiment.groups[0].name
\n treatment_groups = [g.name for g in experiment.groups[1:]]<\/p>\n

results = {}<\/p>\n

for treatment_group in treatment_groups:
\n # \u83b7\u53d6\u8f6c\u5316\u7387\u6570\u636e
\n control_stats = group_stats[control_group]
\n treatment_stats = group_stats[treatment_group]<\/p>\n

# \u68c0\u67e5\u662f\u5426\u6709\u8db3\u591f\u7684\u6570\u636e
\n if control_stats.user_count == 0 or treatment_stats.user_count == 0:
\n results[treatment_group] = {
\n 'significant': False,
\n 'p_value': 1.0,
\n 'effect_size': 0.0,
\n 'error': 'Insufficient data'
\n }
\n continue<\/p>\n

# \u63d0\u53d6\u4e3b\u8981\u6307\u6807（\u5047\u8bbe\u7b2c\u4e00\u4e2a\u6307\u6807\u662f\u4e3b\u8981\u6307\u6807）
\n primary_metric = experiment.metrics[0] if experiment.metrics else 'conversion'<\/p>\n

# \u83b7\u53d6\u6307\u6807\u503c
\n control_values = control_stats.metric_values.get(primary_metric, [])
\n treatment_values = treatment_stats.metric_values.get(primary_metric, [])<\/p>\n

if not control_values or not treatment_values:
\n # \u5982\u679c\u6ca1\u6709\u8fde\u7eed\u503c\u6307\u6807，\u4f7f\u7528\u4e8c\u9879\u5206\u5e03\u68c0\u9a8c
\n control_conversions = control_stats.event_counts.get('conversion', 0)
\n treatment_conversions = treatment_stats.event_counts.get('conversion', 0)<\/p>\n

control_non_conversions = control_stats.user_count – control_conversions
\n treatment_non_conversions = treatment_stats.user_count – treatment_conversions<\/p>\n

# \u5361\u65b9\u68c0\u9a8c
\n from scipy.stats import chi2_contingency
\n contingency_table = [
\n [control_conversions, control_non_conversions],
\n [treatment_conversions, treatment_non_conversions]
\n ]<\/p>\n

chi2, p_value, dof, expected = chi2_contingency(contingency_table)<\/p>\n

# \u8ba1\u7b97\u6548\u5e94\u5927\u5c0f
\n control_rate = control_conversions \/ control_stats.user_count if control_stats.user_count > 0 else 0
\n treatment_rate = treatment_conversions \/ treatment_stats.user_count if treatment_stats.user_count > 0 else 0
\n effect_size = treatment_rate – control_rate<\/p>\n

results[treatment_group] = {
\n 'significant': p_value < 0.05,
\n 'p_value': float(p_value),
\n 'effect_size': float(effect_size),
\n 'relative_improvement': float(effect_size \/ control_rate) if control_rate > 0 else 0.0,
\n 'test_type': 'chi_square'
\n }
\n else:
\n # t\u68c0\u9a8c（\u8fde\u7eed\u503c\u6307\u6807）
\n from scipy.stats import ttest_ind<\/p>\n

t_stat, p_value = ttest_ind(control_values, treatment_values, equal_var=False)<\/p>\n

# \u8ba1\u7b97\u6548\u5e94\u5927\u5c0f（Cohen's d）
\n control_mean = np.mean(control_values)
\n treatment_mean = np.mean(treatment_values)
\n control_std = np.std(control_values, ddof=1)
\n treatment_std = np.std(treatment_values, ddof=1)<\/p>\n

pooled_std = np.sqrt((control_std**2 + treatment_std**2) \/ 2)
\n effect_size = (treatment_mean – control_mean) \/ pooled_std if pooled_std > 0 else 0<\/p>\n

results[treatment_group] = {
\n 'significant': p_value < 0.05,
\n 'p_value': float(p_value),
\n 'effect_size': float(effect_size),
\n 'relative_improvement': float((treatment_mean – control_mean) \/ control_mean) if control_mean > 0 else 0.0,
\n 'test_type': 't_test'
\n }<\/p>\n

return results<\/p>\n

def _format_stats(self, stats: ExperimentStats) -> Dict[str, Any]:
\n """\u683c\u5f0f\u5316\u7edf\u8ba1\u4fe1\u606f"""
\n return {
\n 'experiment_name': stats.experiment_name,
\n 'start_time': stats.start_time.isoformat(),
\n 'end_time': stats.end_time.isoformat() if stats.end_time else None,
\n 'total_users': stats.total_users,
\n 'total_events': stats.total_events,
\n 'groups': {
\n group_name: {
\n 'user_count': group_stats.user_count,
\n 'event_counts': group_stats.event_counts,
\n 'conversions': group_stats.conversions,
\n 'metric_means': {
\n metric: np.mean(values) if values else 0
\n for metric, values in group_stats.metric_values.items()
\n },
\n 'metric_stds': {
\n metric: np.std(values) if values else 0
\n for metric, values in group_stats.metric_values.items()
\n }
\n }
\n for group_name, group_stats in stats.group_stats.items()
\n },
\n 'significance_test': stats.significance_test,
\n 'recommendation': self._generate_recommendation(stats)
\n }<\/p>\n

def _generate_recommendation(self, stats: ExperimentStats) -> str:
\n """\u751f\u6210\u5b9e\u9a8c\u5efa\u8bae"""
\n if not stats.significance_test:
\n return "Insufficient data for recommendation"<\/p>\n

# \u68c0\u67e5\u662f\u5426\u6709\u663e\u8457\u63d0\u5347\u7684\u5206\u7ec4
\n best_group = None
\n best_improvement = 0<\/p>\n

for treatment_group, test_result in stats.significance_test.items():
\n if test_result.get('significant', False):
\n improvement = test_result.get('relative_improvement', 0)
\n if improvement > best_improvement:
\n best_improvement = improvement
\n best_group = treatment_group<\/p>\n

if best_group and best_improvement > 0:
\n return f"Recommend implementing {best_group} (improvement: {best_improvement:.2%})"
\n elif best_group and best_improvement < 0:
\n return f"Warning: {best_group} performs worse than control"
\n else:
\n return "No significant difference detected"<\/p>\n

def _validate_experiment(self, experiment: Experiment) -> bool:
\n """\u9a8c\u8bc1\u5b9e\u9a8c\u914d\u7f6e"""
\n if not experiment.name:
\n self.logger.error("\u5b9e\u9a8c\u540d\u79f0\u4e0d\u80fd\u4e3a\u7a7a")
\n return False<\/p>\n

if not experiment.groups:
\n self.logger.error("\u5b9e\u9a8c\u5fc5\u987b\u5305\u542b\u81f3\u5c11\u4e00\u4e2a\u5206\u7ec4")
\n return False<\/p>\n

# \u68c0\u67e5\u5206\u7ec4\u540d\u79f0\u662f\u5426\u552f\u4e00
\n group_names = [g.name for g in experiment.groups]
\n if len(group_names) != len(set(group_names)):
\n self.logger.error("\u5206\u7ec4\u540d\u79f0\u5fc5\u987b\u552f\u4e00")
\n return False<\/p>\n

# \u68c0\u67e5\u6743\u91cd\u662f\u5426\u5408\u7406
\n total_weight = sum(g.weight for g in experiment.groups)
\n if total_weight <= 0:
\n self.logger.error("\u5206\u7ec4\u6743\u91cd\u603b\u548c\u5fc5\u987b\u5927\u4e8e0")
\n return False<\/p>\n

# \u68c0\u67e5\u91c7\u6837\u7387
\n if not 0 <= experiment.sample_rate <= 1:
\n self.logger.error("\u91c7\u6837\u7387\u5fc5\u987b\u57280\u52301\u4e4b\u95f4")
\n return False<\/p>\n

return True<\/p>\n

def _save_experiments(self):
\n """\u4fdd\u5b58\u5b9e\u9a8c\u914d\u7f6e"""
\n # \u5b9e\u9645\u5b9e\u73b0\u4e2d\u5e94\u4fdd\u5b58\u5230\u6570\u636e\u5e93\u6216\u914d\u7f6e\u6587\u4ef6
\n experiments_data = []
\n for experiment in self.experiments.values():
\n experiments_data.append({
\n 'name': experiment.name,
\n 'description': experiment.description,
\n 'status': experiment.status.value,
\n 'groups': [
\n {
\n 'name': g.name,
\n 'weight': g.weight,
\n 'model_name': g.model_name,
\n 'model_version': g.model_version,
\n 'parameters': g.parameters,
\n 'description': g.description
\n }
\n for g in experiment.groups
\n ],
\n 'assignment_algorithm': experiment.assignment_algorithm.value,
\n 'target_users': experiment.target_users,
\n 'sample_rate': experiment.sample_rate,
\n 'metrics': experiment.metrics,
\n 'hypotheses': experiment.hypotheses
\n })<\/p>\n

# \u4fdd\u5b58\u5230\u6587\u4ef6（\u793a\u4f8b）
\n config_file = self.config.get('experiments_file', 'experiments.json')
\n try:
\n with open(config_file, 'w') as f:
\n json.dump(experiments_data, f, indent=2, ensure_ascii=False)
\n except Exception as e:
\n self.logger.error(f"\u4fdd\u5b58\u5b9e\u9a8c\u914d\u7f6e\u5931\u8d25: {e}")<\/p>\n

# ==================== \u5b9e\u9a8c\u76d1\u63a7\u9762\u677f ====================
\nclass ExperimentDashboard:
\n """\u5b9e\u9a8c\u76d1\u63a7\u9762\u677f"""<\/p>\n

def __init__(self, ab_test_manager: ABTestManager):
\n self.ab_test_manager = ab_test_manager
\n self.logger = logging.getLogger("ExperimentDashboard")<\/p>\n

async def get_dashboard_data(self) -> Dict[str, Any]:
\n """\u83b7\u53d6\u9762\u677f\u6570\u636e"""
\n experiments = self.ab_test_manager.list_experiments()<\/p>\n

dashboard_data = {
\n 'timestamp': datetime.now().isoformat(),
\n 'total_experiments': len(experiments),
\n 'running_experiments': sum(1 for e in experiments if e['status'] == 'running'),
\n 'experiments': []
\n }<\/p>\n

for experiment in experiments:
\n experiment_name = experiment['name']
\n stats = self.ab_test_manager.get_experiment_stats(experiment_name)<\/p>\n

dashboard_data['experiments'].append({
\n 'info': experiment,
\n 'stats': stats
\n })<\/p>\n

return dashboard_data<\/p>\n

def generate_report(self, experiment_name: str) -> Dict[str, Any]:
\n """\u751f\u6210\u5b9e\u9a8c\u62a5\u544a"""
\n stats = self.ab_test_manager.get_experiment_stats(experiment_name)
\n if not stats:
\n return {'error': 'Experiment not found'}<\/p>\n

experiment = self.ab_test_manager.experiments.get(experiment_name)<\/p>\n

report = {
\n 'experiment_name': experiment_name,
\n 'report_date': datetime.now().isoformat(),
\n 'experiment_info': {
\n 'description': experiment.description if experiment else None,
\n 'hypotheses': experiment.hypotheses if experiment else None,
\n 'duration_days': None
\n },
\n 'executive_summary': self._generate_executive_summary(stats),
\n 'methodology': {
\n 'assignment_algorithm': experiment.assignment_algorithm.value if experiment else 'unknown',
\n 'sample_rate': experiment.sample_rate if experiment else 1.0,
\n 'target_users': experiment.target_users if experiment else None
\n },
\n 'results': self._format_results_for_report(stats),
\n 'conclusions': self._generate_conclusions(stats),
\n 'recommendations': stats.get('recommendation', 'No recommendation')
\n }<\/p>\n

if experiment and experiment.start_time:
\n duration = (datetime.now() – experiment.start_time).days
\n report['experiment_info']['duration_days'] = duration<\/p>\n

return report<\/p>\n

def _generate_executive_summary(self, stats: Dict[str, Any]) -> str:
\n """\u751f\u6210\u6267\u884c\u6458\u8981"""
\n total_users = stats.get('total_users', 0)
\n total_events = stats.get('total_events', 0)<\/p>\n

summary = f"""
\n Experiment: {stats.get('experiment_name')}
\n Total Users: {total_users}
\n Total Events: {total_events}
\n """<\/p>\n

significance_test = stats.get('significance_test')
\n if significance_test:
\n summary += "\\\\nSignificance Test Results:\\\\n"
\n for group, result in significance_test.items():
\n if result.get('significant', False):
\n improvement = result.get('relative_improvement', 0)
\n summary += f" – {group}: Significant improvement of {improvement:.2%}\\\\n"
\n else:
\n summary += f" – {group}: No significant difference\\\\n"<\/p>\n

return summary<\/p>\n

def _format_results_for_report(self, stats: Dict[str, Any]) -> Dict[str, Any]:
\n """\u683c\u5f0f\u5316\u7ed3\u679c\u7528\u4e8e\u62a5\u544a"""
\n formatted_results = {
\n 'overall_metrics': {
\n 'total_users': stats.get('total_users', 0),
\n 'total_events': stats.get('total_events', 0)
\n },
\n 'group_performance': {}
\n }<\/p>\n

groups = stats.get('groups', {})
\n for group_name, group_stats in groups.items():
\n formatted_results['group_performance'][group_name] = {
\n 'user_count': group_stats.get('user_count', 0),
\n 'conversion_rate': group_stats.get('conversions', {}).get('conversion', 0),
\n 'key_metrics': {
\n metric: {
\n 'mean': group_stats.get('metric_means', {}).get(metric, 0),
\n 'std': group_stats.get('metric_stds', {}).get(metric, 0)
\n }
\n for metric in group_stats.get('metric_means', {}).keys()
\n }
\n }<\/p>\n

return formatted_results<\/p>\n

def _generate_conclusions(self, stats: Dict[str, Any]) -> List[str]:
\n """\u751f\u6210\u7ed3\u8bba"""
\n conclusions = []<\/p>\n

significance_test = stats.get('significance_test')
\n if not significance_test:
\n conclusions.append("\u5b9e\u9a8c\u6570\u636e\u4e0d\u8db3，\u65e0\u6cd5\u5f97\u51fa\u7ed3\u8bba")
\n return conclusions<\/p>\n

for group, result in significance_test.items():
\n if result.get('significant', False):
\n improvement = result.get('relative_improvement', 0)
\n p_value = result.get('p_value', 1.0)
\n conclusions.append(
\n f"\u5206\u7ec4 {group} \u76f8\u5bf9\u4e8e\u63a7\u5236\u7ec4\u6709\u663e\u8457\u6539\u5584 (p={p_value:.4f}, \u63d0\u5347={improvement:.2%})"
\n )
\n else:
\n p_value = result.get('p_value', 1.0)
\n conclusions.append(
\n f"\u5206\u7ec4 {group} \u76f8\u5bf9\u4e8e\u63a7\u5236\u7ec4\u6ca1\u6709\u663e\u8457\u5dee\u5f02 (p={p_value:.4f})"
\n )<\/p>\n

return conclusions<\/p>\n

# ==================== \u914d\u7f6e\u793a\u4f8b ====================
\ndef create_sample_config() -> Dict[str, Any]:
\n """\u521b\u5efa\u793a\u4f8b\u914d\u7f6e"""
\n return {
\n 'experiments_file': 'experiments.json',
\n 'experiments': [
\n {
\n 'name': 'recommendation_model_v2',
\n 'description': '\u6d4b\u8bd5\u65b0\u7248\u63a8\u8350\u6a21\u578b\u7684\u6548\u679c',
\n 'status': 'running',
\n 'groups': [
\n {
\n 'name': 'control',
\n 'weight': 0.5,
\n 'model_name': 'recommendation',
\n 'model_version': 'v1',
\n 'description': '\u5f53\u524d\u751f\u4ea7\u6a21\u578b'
\n },
\n {
\n 'name': 'treatment_v2',
\n 'weight': 0.3,
\n 'model_name': 'recommendation',
\n 'model_version': 'v2',
\n 'description': '\u65b0\u7248\u63a8\u8350\u6a21\u578b'
\n },
\n {
\n 'name': 'treatment_v3',
\n 'weight': 0.2,
\n 'model_name': 'recommendation',
\n 'model_version': 'v3',
\n 'description': '\u5b9e\u9a8c\u6027\u63a8\u8350\u6a21\u578b'
\n }
\n ],
\n 'assignment_algorithm': 'hash_based',
\n 'target_users': None, # \u6240\u6709\u7528\u6237
\n 'sample_rate': 0.1, # 10%\u6d41\u91cf
\n 'metrics': ['click_rate', 'purchase_rate', 'revenue'],
\n 'hypotheses': '\u65b0\u7248\u63a8\u8350\u6a21\u578b\u80fd\u63d0\u9ad8\u7528\u6237\u70b9\u51fb\u7387\u548c\u8d2d\u4e70\u7387'
\n },
\n {
\n 'name': 'pricing_strategy_test',
\n 'description': '\u6d4b\u8bd5\u4e0d\u540c\u5b9a\u4ef7\u7b56\u7565',
\n 'status': 'draft',
\n 'groups': [
\n {
\n 'name': 'standard_pricing',
\n 'weight': 0.33,
\n 'parameters': {
\n 'price_multiplier': 1.0,
\n 'discount_rate': 0.0
\n },
\n 'description': '\u6807\u51c6\u5b9a\u4ef7'
\n },
\n {
\n 'name': 'discounted_pricing',
\n 'weight': 0.33,
\n 'parameters': {
\n 'price_multiplier': 0.9,
\n 'discount_rate': 0.1
\n },
\n 'description': '9\u6298\u5b9a\u4ef7'
\n },
\n {
\n 'name': 'premium_pricing',
\n 'weight': 0.34,
\n 'parameters': {
\n 'price_multiplier': 1.1,
\n 'premium_features': True
\n },
\n 'description': '\u9ad8\u7aef\u5b9a\u4ef7'
\n }
\n ],
\n 'assignment_algorithm': 'random',
\n 'sample_rate': 0.05, # 5%\u6d41\u91cf
\n 'metrics': ['conversion_rate', 'average_order_value', 'customer_satisfaction'],
\n 'hypotheses': '\u9002\u5ea6\u7684\u6298\u6263\u80fd\u63d0\u9ad8\u8f6c\u5316\u7387\u800c\u4e0d\u663e\u8457\u964d\u4f4e\u5ba2\u5355\u4ef7'
\n }
\n ]
\n }<\/p>\n

if __name__ == "__main__":
\n # \u793a\u4f8b\u7528\u6cd5
\n config = create_sample_config()<\/p>\n

# \u521b\u5efaA\/B\u6d4b\u8bd5\u7ba1\u7406\u5668
\n ab_test_manager = ABTestManager(config)<\/p>\n

# \u542f\u52a8\u5b9e\u9a8c
\n ab_test_manager.start_experiment('recommendation_model_v2')<\/p>\n

# \u6a21\u62df\u7528\u6237\u5206\u914d
\n for i in range(1000):
\n user_id = f"user_{i}"
\n assignment = ab_test_manager.assign_experiment(
\n user_id=user_id,
\n experiment_name='recommendation_model_v2'
\n )<\/p>\n

if assignment:
\n print(f"\u7528\u6237 {user_id} \u5206\u914d\u5230\u5206\u7ec4 {assignment['group_name']}")<\/p>\n

# \u6a21\u62df\u7528\u6237\u884c\u4e3a
\n if random.random() < 0.1: # 10%\u7684\u70b9\u51fb\u7387
\n ab_test_manager.log_event(
\n request_id=assignment['assignment_id'],
\n user_id=user_id,
\n experiment_name='recommendation_model_v2',
\n group_name=assignment['group_name'],
\n action='click',
\n metadata={'timestamp': datetime.now().isoformat()}
\n )<\/p>\n

if random.random() < 0.2: # 20%\u7684\u8d2d\u4e70\u8f6c\u5316\u7387
\n ab_test_manager.log_event(
\n request_id=assignment['assignment_id'],
\n user_id=user_id,
\n experiment_name='recommendation_model_v2',
\n group_name=assignment['group_name'],
\n action='purchase',
\n metadata={
\n 'amount': random.uniform(10, 100),
\n 'timestamp': datetime.now().isoformat()
\n }
\n )<\/p>\n

# \u83b7\u53d6\u5b9e\u9a8c\u7edf\u8ba1
\n stats = ab_test_manager.get_experiment_stats('recommendation_model_v2')
\n print(json.dumps(stats, indent=2, ensure_ascii=False))<\/p>\n

# \u521b\u5efa\u76d1\u63a7\u9762\u677f
\n dashboard = ExperimentDashboard(ab_test_manager)
\n report = dashboard.generate_report('recommendation_model_v2')
\n print(json.dumps(report, indent=2, ensure_ascii=False)) <\/p>\n

\u00a0\u7bc7\u5e45\u9650\u5236\u4e0b\u9762\u5c31\u53ea\u80fd\u7ed9\u5927\u5bb6\u5c55\u793a\u5c0f\u518c\u90e8\u5206\u5185\u5bb9\u4e86\u3002\u6574\u7406\u4e86\u4e00\u4efd\u6838\u5fc3\u9762\u8bd5\u7b14\u8bb0\u5305\u62ec\u4e86：Java\u9762\u8bd5\u3001Spring\u3001JVM\u3001MyBatis\u3001Redis\u3001MySQL\u3001\u5e76\u53d1\u7f16\u7a0b\u3001\u5fae\u670d\u52a1\u3001Linux\u3001Springboot\u3001SpringCloud\u3001MQ\u3001Kafc<\/p>\n

\u9700\u8981\u5168\u5957\u9762\u8bd5\u7b14\u8bb0\u53ca\u7b54\u6848 \u3010\u70b9\u51fb\u6b64\u5904\u5373\u53ef\/\u514d\u8d39\u83b7\u53d6\u3011\u200b\u200b\u200b<\/span><\/p>\n<\/p>\n

2.2 \u5b9e\u9a8c\u5206\u6790\u4e0e\u76d1\u63a7<\/h4>\n

python<\/p>\n<\/p>\n

\u590d\u5236<\/p>\n<\/p>\n

\u4e0b\u8f7d<\/p>\n

"""
\n\u5b9e\u9a8c\u5206\u6790\u4e0e\u76d1\u63a7\u7cfb\u7edf
\n"""
\nimport asyncio
\nimport json
\nimport time
\nfrom datetime import datetime, timedelta
\nfrom typing import Dict, List, Optional, Any, Tuple
\nimport numpy as np
\nimport pandas as pd
\nfrom scipy import stats
\nimport plotly.graph_objs as go
\nimport plotly.offline as pyo
\nfrom plotly.subplots import make_subplots
\nimport matplotlib.pyplot as plt
\nimport seaborn as sns
\nfrom sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, Text
\nfrom sqlalchemy.ext.declarative import declarative_base
\nfrom sqlalchemy.orm import sessionmaker<\/p>\n

# ==================== \u6570\u636e\u5e93\u6a21\u578b ====================
\nBase = declarative_base()<\/p>\n

class ExperimentRecord(Base):
\n """\u5b9e\u9a8c\u8bb0\u5f55\u8868"""
\n __tablename__ = 'experiment_records'<\/p>\n

id = Column(Integer, primary_key=True)
\n experiment_name = Column(String(255), nullable=False, index=True)
\n user_id = Column(String(255), nullable=False, index=True)
\n group_name = Column(String(100), nullable=False)
\n event_type = Column(String(100), nullable=False)
\n event_value = Column(Float, nullable=True)
\n event_metadata = Column(Text, nullable=True) # JSON\u683c\u5f0f
\n timestamp = Column(DateTime, nullable=False, index=True)
\n created_at = Column(DateTime, default=datetime.now)<\/p>\n

class ExperimentAnalysis:
\n """\u5b9e\u9a8c\u5206\u6790\u5668"""<\/p>\n

def __init__(self, db_url: str = "sqlite:\/\/\/experiments.db"):
\n """\u521d\u59cb\u5316\u5206\u6790\u5668"""
\n self.engine = create_engine(db_url)
\n Base.metadata.create_all(self.engine)
\n self.Session = sessionmaker(bind=self.engine)<\/p>\n

# \u7edf\u8ba1\u6d4b\u8bd5\u914d\u7f6e
\n self.alpha = 0.05 # \u663e\u8457\u6027\u6c34\u5e73
\n self.min_sample_size = 30 # \u6700\u5c0f\u6837\u672c\u91cf
\n self.power = 0.8 # \u7edf\u8ba1\u529f\u6548<\/p>\n

def record_event(self, event_data: Dict[str, Any]):
\n """\u8bb0\u5f55\u4e8b\u4ef6\u5230\u6570\u636e\u5e93"""
\n session = self.Session()
\n try:
\n record = ExperimentRecord(
\n experiment_name=event_data['experiment_name'],
\n user_id=event_data['user_id'],
\n group_name=event_data['group_name'],
\n event_type=event_data['event_type'],
\n event_value=event_data.get('event_value'),
\n event_metadata=json.dumps(event_data.get('metadata', {})),
\n timestamp=event_data.get('timestamp', datetime.now())
\n )
\n session.add(record)
\n session.commit()
\n except Exception as e:
\n session.rollback()
\n raise e
\n finally:
\n session.close()<\/p>\n

async def analyze_experiment(
\n self,
\n experiment_name: str,
\n start_date: Optional[datetime] = None,
\n end_date: Optional[datetime] = None
\n ) -> Dict[str, Any]:
\n """\u5206\u6790\u5b9e\u9a8c\u6570\u636e"""
\n session = self.Session()<\/p>\n

try:
\n # \u6784\u5efa\u67e5\u8be2
\n query = session.query(ExperimentRecord).filter(
\n ExperimentRecord.experiment_name == experiment_name
\n )<\/p>\n

if start_date:
\n query = query.filter(ExperimentRecord.timestamp >= start_date)
\n if end_date:
\n query = query.filter(ExperimentRecord.timestamp <= end_date)<\/p>\n

# \u83b7\u53d6\u6570\u636e
\n records = query.all()<\/p>\n

if not records:
\n return {"error": "No data found for experiment"}<\/p>\n

# \u8f6c\u6362\u4e3aDataFrame
\n data = []
\n for record in records:
\n data.append({
\n 'user_id': record.user_id,
\n 'group_name': record.group_name,
\n 'event_type': record.event_type,
\n 'event_value': record.event_value,
\n 'timestamp': record.timestamp,
\n 'metadata': json.loads(record.event_metadata) if record.event_metadata else {}
\n })<\/p>\n

df = pd.DataFrame(data)<\/p>\n

# \u5206\u6790\u4e0d\u540c\u5206\u7ec4\u7684\u8868\u73b0
\n analysis_result = {
\n 'experiment_name': experiment_name,
\n 'analysis_date': datetime.now().isoformat(),
\n 'time_period': {
\n 'start': start_date.isoformat() if start_date else df['timestamp'].min().isoformat(),
\n 'end': end_date.isoformat() if end_date else df['timestamp'].max().isoformat()
\n },
\n 'summary': self._generate_summary(df),
\n 'group_comparison': await self._compare_groups(df),
\n 'time_series_analysis': self._analyze_time_series(df),
\n 'power_analysis': self._calculate_power_analysis(df),
\n 'sensitivity_analysis': self._sensitivity_analysis(df),
\n 'visualizations': await self._generate_visualizations(df, experiment_name)
\n }<\/p>\n

return analysis_result<\/p>\n

finally:
\n session.close()<\/p>\n

def _generate_summary(self, df: pd.DataFrame) -> Dict[str, Any]:
\n """\u751f\u6210\u6570\u636e\u6458\u8981"""
\n total_users = df['user_id'].nunique()
\n total_events = len(df)<\/p>\n

summary = {
\n 'total_users': int(total_users),
\n 'total_events': int(total_events),
\n 'events_per_user': float(total_events \/ total_users) if total_users > 0 else 0,
\n 'groups': {}
\n }<\/p>\n

# \u6309\u5206\u7ec4\u7edf\u8ba1
\n for group_name, group_df in df.groupby('group_name'):
\n group_users = group_df['user_id'].nunique()
\n group_events = len(group_df)<\/p>\n

# \u4e8b\u4ef6\u7c7b\u578b\u7edf\u8ba1
\n event_counts = group_df['event_type'].value_counts().to_dict()<\/p>\n

# \u6570\u503c\u578b\u4e8b\u4ef6\u7684\u7edf\u8ba1
\n numeric_events = group_df[group_df['event_value'].notnull()]
\n event_stats = {}
\n if not numeric_events.empty:
\n for event_type, event_df in numeric_events.groupby('event_type'):
\n values = event_df['event_value'].astype(float)
\n event_stats[event_type] = {
\n 'count': int(len(values)),
\n 'mean': float(values.mean()),
\n 'std': float(values.std()),
\n 'min': float(values.min()),
\n 'max': float(values.max()),
\n 'median': float(values.median())
\n }<\/p>\n

summary['groups'][group_name] = {
\n 'user_count': int(group_users),
\n 'event_count': int(group_events),
\n 'event_distribution': event_counts,
\n 'event_statistics': event_stats
\n }<\/p>\n

return summary<\/p>\n

async def _compare_groups(self, df: pd.DataFrame) -> Dict[str, Any]:
\n """\u6bd4\u8f83\u4e0d\u540c\u5206\u7ec4\u7684\u8868\u73b0"""
\n # \u8bc6\u522b\u63a7\u5236\u7ec4（\u5047\u8bbe\u7b2c\u4e00\u4e2a\u5206\u7ec4\u662f\u63a7\u5236\u7ec4）
\n groups = df['group_name'].unique()
\n if len(groups) < 2:
\n return {"error": "Need at least 2 groups for comparison"}<\/p>\n

control_group = groups[0]
\n treatment_groups = groups[1:]<\/p>\n

comparisons = {}<\/p>\n

for treatment_group in treatment_groups:
\n comparison = await self._compare_two_groups(
\n df, control_group, treatment_group
\n )
\n comparisons[treatment_group] = comparison<\/p>\n

return {
\n 'control_group': control_group,
\n 'treatment_groups': list(treatment_groups),
\n 'comparisons': comparisons,
\n 'overall_recommendation': self._generate_overall_recommendation(comparisons)
\n }<\/p>\n

async def _compare_two_groups(
\n self,
\n df: pd.DataFrame,
\n control_group: str,
\n treatment_group: str
\n ) -> Dict[str, Any]:
\n """\u6bd4\u8f83\u4e24\u4e2a\u5206\u7ec4"""
\n control_df = df[df['group_name'] == control_group]
\n treatment_df = df[df['group_name'] == treatment_group]<\/p>\n

comparison = {
\n 'control_group': control_group,
\n 'treatment_group': treatment_group,
\n 'user_counts': {
\n 'control': int(control_df['user_id'].nunique()),
\n 'treatment': int(treatment_df['user_id'].nunique())
\n },
\n 'event_comparisons': {}
\n }<\/p>\n

# \u6bd4\u8f83\u6bcf\u79cd\u4e8b\u4ef6\u7c7b\u578b
\n all_event_types = set(control_df['event_type'].unique()) | set(treatment_df['event_type'].unique())<\/p>\n

for event_type in all_event_types:
\n control_events = control_df[control_df['event_type'] == event_type]
\n treatment_events = treatment_df[treatment_df['event_type'] == event_type]<\/p>\n

# \u8ba1\u7b97\u8f6c\u5316\u7387
\n control_users = control_df['user_id'].nunique()
\n treatment_users = treatment_df['user_id'].nunique()<\/p>\n

control_conversions = control_events['user_id'].nunique() if not control_events.empty else 0
\n treatment_conversions = treatment_events['user_id'].nunique() if not treatment_events.empty else 0<\/p>\n

control_rate = control_conversions \/ control_users if control_users > 0 else 0
\n treatment_rate = treatment_conversions \/ treatment_users if treatment_users > 0 else 0<\/p>\n

# \u8ba1\u7b97\u76f8\u5bf9\u63d0\u5347
\n relative_improvement = (treatment_rate – control_rate) \/ control_rate if control_rate > 0 else 0<\/p>\n

# \u663e\u8457\u6027\u68c0\u9a8c（\u6bd4\u4f8b\u68c0\u9a8c）
\n p_value = 1.0
\n if control_conversions > 0 and treatment_conversions > 0:
\n from statsmodels.stats.proportion import proportions_ztest<\/p>\n

count = [treatment_conversions, control_conversions]
\n nobs = [treatment_users, control_users]<\/p>\n

try:
\n z_stat, p_value = proportions_ztest(count, nobs, alternative='larger')
\n except:
\n p_value = 1.0<\/p>\n

# \u6570\u503c\u578b\u4e8b\u4ef6\u7684\u6bd4\u8f83
\n value_comparison = None
\n control_values = control_events['event_value'].dropna()
\n treatment_values = treatment_events['event_value'].dropna()<\/p>\n

if not control_values.empty and not treatment_values.empty:
\n # t\u68c0\u9a8c
\n t_stat, t_p_value = stats.ttest_ind(
\n control_values.astype(float),
\n treatment_values.astype(float),
\n equal_var=False
\n )<\/p>\n

# \u8ba1\u7b97\u6548\u5e94\u5927\u5c0f
\n control_mean = control_values.mean()
\n treatment_mean = treatment_values.mean()
\n control_std = control_values.std()
\n treatment_std = treatment_values.std()<\/p>\n

pooled_std = np.sqrt((control_std**2 + treatment_std**2) \/ 2)
\n cohens_d = (treatment_mean – control_mean) \/ pooled_std if pooled_std > 0 else 0<\/p>\n

value_comparison = {
\n 'control_mean': float(control_mean),
\n 'treatment_mean': float(treatment_mean),
\n 'mean_difference': float(treatment_mean – control_mean),
\n 't_statistic': float(t_stat),
\n 'p_value': float(t_p_value),
\n 'cohens_d': float(cohens_d),
\n 'significant': t_p_value < self.alpha
\n }<\/p>\n

comparison['event_comparisons'][event_type] = {
\n 'conversion_rates': {
\n 'control': float(control_rate),
\n 'treatment': float(treatment_rate)
\n },
\n 'absolute_difference': float(treatment_rate – control_rate),
\n 'relative_improvement': float(relative_improvement),
\n 'p_value': float(p_value),
\n 'significant': p_value < self.alpha,
\n 'value_comparison': value_comparison
\n }<\/p>\n

return comparison<\/p>\n

def _analyze_time_series(self, df: pd.DataFrame) -> Dict[str, Any]:
\n """\u65f6\u95f4\u5e8f\u5217\u5206\u6790"""
\n df = df.copy()
\n df['date'] = df['timestamp'].dt.date<\/p>\n

# \u6309\u65e5\u671f\u548c\u5206\u7ec4\u7edf\u8ba1
\n daily_stats = df.groupby(['date', 'group_name']).agg({
\n 'user_id': 'nunique',
\n 'event_type': 'count'
\n }).reset_index()<\/p>\n

daily_stats.columns = ['date', 'group_name', 'daily_users', 'daily_events']<\/p>\n

# \u8ba1\u7b97\u7d2f\u79ef\u7edf\u8ba1
\n daily_stats['cumulative_users'] = daily_stats.groupby('group_name')['daily_users'].cumsum()
\n daily_stats['cumulative_events'] = daily_stats.groupby('group_name')['daily_events'].cumsum()<\/p>\n

# \u8ba1\u7b97\u6bcf\u65e5\u8f6c\u5316\u7387
\n daily_stats['daily_conversion_rate'] = daily_stats['daily_events'] \/ daily_stats['daily_users']
\n daily_stats['cumulative_conversion_rate'] = daily_stats['cumulative_events'] \/ daily_stats['cumulative_users']<\/p>\n

# \u8f6c\u6362\u4e3a\u5b57\u5178\u683c\u5f0f
\n time_series_data = {}
\n for group_name, group_df in daily_stats.groupby('group_name'):
\n time_series_data[group_name] = {
\n 'dates': [d.isoformat() for d in group_df['date']],
\n 'daily_users': group_df['daily_users'].tolist(),
\n 'daily_events': group_df['daily_events'].tolist(),
\n 'daily_conversion_rates': group_df['daily_conversion_rate'].tolist(),
\n 'cumulative_conversion_rates': group_df['cumulative_conversion_rate'].tolist()
\n }<\/p>\n

return {
\n 'time_series_data': time_series_data,
\n 'trend_analysis': self._analyze_trends(daily_stats)
\n }<\/p>\n

def _analyze_trends(self, daily_stats: pd.DataFrame) -> Dict[str, Any]:
\n """\u5206\u6790\u8d8b\u52bf"""
\n trends = {}<\/p>\n

for group_name, group_df in daily_stats.groupby('group_name'):
\n # \u8ba1\u7b97\u6bcf\u65e5\u8f6c\u5316\u7387\u7684\u8d8b\u52bf
\n dates = pd.to_datetime(group_df['date'])
\n conversion_rates = group_df['daily_conversion_rate'].fillna(0)<\/p>\n

if len(conversion_rates) > 1:
\n # \u7ebf\u6027\u8d8b\u52bf
\n x = np.arange(len(conversion_rates))
\n slope, intercept = np.polyfit(x, conversion_rates, 1)<\/p>\n

# \u8ba1\u7b97\u8d8b\u52bf\u5f3a\u5ea6（R\u00b2）
\n y_pred = slope * x + intercept
\n ss_res = np.sum((conversion_rates – y_pred) ** 2)
\n ss_tot = np.sum((conversion_rates – np.mean(conversion_rates)) ** 2)
\n r_squared = 1 – (ss_res \/ ss_tot) if ss_tot > 0 else 0<\/p>\n

trends[group_name] = {
\n 'slope': float(slope),
\n 'intercept': float(intercept),
\n 'r_squared': float(r_squared),
\n 'trend_direction': 'increasing' if slope > 0 else 'decreasing',
\n 'trend_strength': 'strong' if abs(slope) > 0.01 else 'weak'
\n }
\n else:
\n trends[group_name] = {
\n 'error': 'Insufficient data for trend analysis'
\n }<\/p>\n

return trends<\/p>\n

def _calculate_power_analysis(self, df: pd.DataFrame) -> Dict[str, Any]:
\n """\u8ba1\u7b97\u7edf\u8ba1\u529f\u6548\u5206\u6790"""
\n from statsmodels.stats.power import TTestIndPower, NormalIndPower<\/p>\n

power_analysis = {}<\/p>\n

for group_name, group_df in df.groupby('group_name'):
\n # \u8ba1\u7b97MDE（\u6700\u5c0f\u53ef\u68c0\u6d4b\u6548\u5e94）
\n users = group_df['user_id'].nunique()<\/p>\n

if users >= 2:
\n # \u8ba1\u7b97\u5f53\u524d\u6548\u5e94\u7684\u53d8\u5f02\u6027
\n conversion_rates = []
\n for event_type, event_df in group_df.groupby('event_type'):
\n event_users = event_df['user_id'].nunique()
\n conversion_rate = event_users \/ users if users > 0 else 0
\n conversion_rates.append(conversion_rate)<\/p>\n

if conversion_rates:
\n effect_size = np.std(conversion_rates)<\/p>\n

# \u8ba1\u7b97\u9700\u8981\u7684\u6837\u672c\u91cf
\n power_analysis_obj = TTestIndPower()
\n required_n = power_analysis_obj.solve_power(
\n effect_size=effect_size,
\n power=self.power,
\n alpha=self.alpha,
\n ratio=1.0
\n )<\/p>\n

power_analysis[group_name] = {
\n 'current_sample_size': int(users),
\n 'effect_size_variability': float(effect_size),
\n 'required_sample_size': float(required_n) if not np.isnan(required_n) else None,
\n 'sufficient_power': users >= required_n if required_n else False
\n }
\n else:
\n power_analysis[group_name] = {
\n 'error': 'No conversion data available'
\n }
\n else:
\n power_analysis[group_name] = {
\n 'error': 'Insufficient users for power analysis'
\n }<\/p>\n

return power_analysis<\/p>\n

def _sensitivity_analysis(self, df: pd.DataFrame) -> Dict[str, Any]:
\n """\u654f\u611f\u6027\u5206\u6790"""
\n sensitivity = {}<\/p>\n

# \u5206\u6790\u4e0d\u540c\u65f6\u95f4\u6bb5\u7684\u8868\u73b0
\n if len(df) > 0:
\n df = df.copy()
\n df['week'] = df['timestamp'].dt.isocalendar().week<\/p>\n

weekly_analysis = {}
\n for week, week_df in df.groupby('week'):
\n week_summary = self._generate_summary(week_df)
\n weekly_analysis[f'week_{week}'] = week_summary<\/p>\n

sensitivity['weekly_analysis'] = weekly_analysis<\/p>\n

# \u5206\u6790\u4e0d\u540c\u7528\u6237\u5b50\u96c6
\n user_ids = df['user_id'].unique()
\n if len(user_ids) > 100:
\n # \u968f\u673a\u62bd\u6837\u5206\u6790
\n np.random.seed(42)
\n sample_sizes = [0.5, 0.7, 0.9] # \u4e0d\u540c\u91c7\u6837\u6bd4\u4f8b<\/p>\n

sampling_analysis = {}
\n for sample_size in sample_sizes:
\n sample_users = np.random.choice(
\n user_ids,
\n size=int(len(user_ids) * sample_size),
\n replace=False
\n )
\n sample_df = df[df['user_id'].isin(sample_users)]
\n sample_summary = self._generate_summary(sample_df)
\n sampling_analysis[f'sample_{int(sample_size*100)}_percent'] = sample_summary<\/p>\n

sensitivity['sampling_analysis'] = sampling_analysis<\/p>\n

return sensitivity<\/p>\n

async def _generate_visualizations(
\n self,
\n df: pd.DataFrame,
\n experiment_name: str
\n ) -> Dict[str, str]:
\n """\u751f\u6210\u53ef\u89c6\u5316\u56fe\u8868"""
\n visualizations = {}<\/p>\n

try:
\n # 1. \u8f6c\u5316\u7387\u5bf9\u6bd4\u56fe
\n fig1 = self._create_conversion_rate_chart(df)
\n visualizations['conversion_rate_chart'] = fig1.to_html(full_html=False)<\/p>\n

# 2. \u65f6\u95f4\u5e8f\u5217\u56fe
\n fig2 = self._create_time_series_chart(df)
\n visualizations['time_series_chart'] = fig2.to_html(full_html=False)<\/p>\n

# 3. \u5206\u5e03\u5bf9\u6bd4\u56fe
\n fig3 = self._create_distribution_chart(df)
\n visualizations['distribution_chart'] = fig3.to_html(full_html=False)<\/p>\n

# 4. \u7d2f\u79ef\u6548\u679c\u56fe
\n fig4 = self._create_cumulative_effect_chart(df)
\n visualizations['cumulative_effect_chart'] = fig4.to_html(full_html=False)<\/p>\n

except Exception as e:
\n visualizations['error'] = str(e)<\/p>\n

return visualizations<\/p>\n

def _create_conversion_rate_chart(self, df: pd.DataFrame) -> go.Figure:
\n """\u521b\u5efa\u8f6c\u5316\u7387\u5bf9\u6bd4\u56fe"""
\n conversion_rates = []
\n groups = []<\/p>\n

for group_name, group_df in df.groupby('group_name'):
\n users = group_df['user_id'].nunique()
\n conversions = group_df[group_df['event_type'] == 'conversion']
\n conversion_count = conversions['user_id'].nunique() if not conversions.empty else 0<\/p>\n

rate = conversion_count \/ users if users > 0 else 0
\n conversion_rates.append(rate * 100) # \u8f6c\u6362\u4e3a\u767e\u5206\u6bd4
\n groups.append(group_name)<\/p>\n

fig = go.Figure(data=[
\n go.Bar(
\n x=groups,
\n y=conversion_rates,
\n text=[f'{rate:.2f}%' for rate in conversion_rates],
\n textposition='auto',
\n marker_color='steelblue'
\n )
\n ])<\/p>\n

fig.update_layout(
\n title='Conversion Rates by Group',
\n xaxis_title='Group',
\n yaxis_title='Conversion Rate (%)',
\n template='plotly_white'
\n )<\/p>\n

return fig<\/p>\n

def _create_time_series_chart(self, df: pd.DataFrame) -> go.Figure:
\n """\u521b\u5efa\u65f6\u95f4\u5e8f\u5217\u56fe"""
\n df = df.copy()
\n df['date'] = df['timestamp'].dt.date<\/p>\n

fig = go.Figure()<\/p>\n

for group_name, group_df in df.groupby('group_name'):
\n daily_stats = group_df.groupby('date').agg({
\n 'user_id': 'nunique',
\n 'event_type': 'count'
\n }).reset_index()<\/p>\n

if not daily_stats.empty:
\n daily_stats['conversion_rate'] = (
\n daily_stats['event_type'] \/ daily_stats['user_id']
\n ) * 100<\/p>\n

fig.add_trace(go.Scatter(
\n x=daily_stats['date'],
\n y=daily_stats['conversion_rate'],
\n mode='lines+markers',
\n name=group_name,
\n line=dict(width=2)
\n ))<\/p>\n

fig.update_layout(
\n title='Daily Conversion Rate Trends',
\n xaxis_title='Date',
\n yaxis_title='Conversion Rate (%)',
\n template='plotly_white',
\n hovermode='x unified'
\n )<\/p>\n

return fig<\/p>\n

def _create_distribution_chart(self, df: pd.DataFrame) -> go.Figure:
\n """\u521b\u5efa\u5206\u5e03\u5bf9\u6bd4\u56fe"""
\n # \u63d0\u53d6\u6570\u503c\u578b\u4e8b\u4ef6
\n numeric_df = df[df['event_value'].notnull()].copy()<\/p>\n

if numeric_df.empty:
\n # \u5982\u679c\u6ca1\u6709\u6570\u503c\u578b\u4e8b\u4ef6，\u521b\u5efa\u7a7a\u7684\u56fe\u8868
\n fig = go.Figure()
\n fig.update_layout(
\n title='No numeric data available for distribution analysis',
\n template='plotly_white'
\n )
\n return fig<\/p>\n

# \u83b7\u53d6\u4e3b\u8981\u7684\u4e8b\u4ef6\u7c7b\u578b
\n main_event_type = numeric_df['event_type'].value_counts().index[0]
\n event_df = numeric_df[numeric_df['event_type'] == main_event_type]<\/p>\n

fig = go.Figure()<\/p>\n

for group_name, group_df in event_df.groupby('group_name'):
\n fig.add_trace(go.Violin(
\n y=group_df['event_value'],
\n name=group_name,
\n box_visible=True,
\n meanline_visible=True
\n ))<\/p>\n

fig.update_layout(
\n title=f'Distribution of {main_event_type} by Group',
\n yaxis_title='Value',
\n template='plotly_white'
\n )<\/p>\n

return fig<\/p>\n

def _create_cumulative_effect_chart(self, df: pd.DataFrame) -> go.Figure:
\n """\u521b\u5efa\u7d2f\u79ef\u6548\u679c\u56fe"""
\n df = df.copy()
\n df['date'] = df['timestamp'].dt.date<\/p>\n

# \u6309\u65e5\u671f\u548c\u5206\u7ec4\u8ba1\u7b97\u7d2f\u79ef\u8f6c\u5316\u7387
\n cumulative_data = []<\/p>\n

for group_name, group_df in df.groupby('group_name'):
\n group_df = group_df.sort_values('date')<\/p>\n

unique_dates = group_df['date'].unique()
\n cumulative_users = 0
\n cumulative_conversions = 0<\/p>\n

for date in unique_dates:
\n date_df = group_df[group_df['date'] == date]
\n daily_users = date_df['user_id'].nunique()
\n daily_conversions = len(date_df[date_df['event_type'] == 'conversion'])<\/p>\n

cumulative_users += daily_users
\n cumulative_conversions += daily_conversions<\/p>\n

cumulative_rate = (cumulative_conversions \/ cumulative_users * 100
\n if cumulative_users > 0 else 0)<\/p>\n

cumulative_data.append({
\n 'date': date,
\n 'group': group_name,
\n 'cumulative_rate': cumulative_rate
\n })<\/p>\n

cumulative_df = pd.DataFrame(cumulative_data)<\/p>\n

fig = go.Figure()<\/p>\n

for group_name, group_df in cumulative_df.groupby('group'):
\n fig.add_trace(go.Scatter(
\n x=group_df['date'],
\n y=group_df['cumulative_rate'],
\n mode='lines',
\n name=group_name,
\n line=dict(width=3)
\n ))<\/p>\n

fig.update_layout(
\n title='Cumulative Conversion Rate Over Time',
\n xaxis_title='Date',
\n yaxis_title='Cumulative Conversion Rate (%)',
\n template='plotly_white',
\n hovermode='x unified'
\n )<\/p>\n

return fig<\/p>\n

def _generate_overall_recommendation(
\n self,
\n comparisons: Dict[str, Any]
\n ) -> Dict[str, Any]:
\n """\u751f\u6210\u603b\u4f53\u63a8\u8350"""
\n if not comparisons:
\n return {"decision": "no_data", "reason": "No comparison data available"}<\/p>\n

# \u68c0\u67e5\u662f\u5426\u6709\u663e\u8457\u63d0\u5347\u7684\u5206\u7ec4
\n best_group = None
\n best_improvement = 0
\n best_p_value = 1.0<\/p>\n

for treatment_group, comparison in comparisons.items():
\n event_comparisons = comparison.get('event_comparisons', {})<\/p>\n

for event_type, event_comp in event_comparisons.items():
\n if event_comp.get('significant', False):
\n improvement = event_comp.get('relative_improvement', 0)
\n p_value = event_comp.get('p_value', 1.0)<\/p>\n

if improvement > best_improvement:
\n best_improvement = improvement
\n best_p_value = p_value
\n best_group = treatment_group<\/p>\n

if best_group and best_improvement > 0:
\n return {
\n "decision": "implement",
\n "recommended_group": best_group,
\n "expected_improvement": f"{best_improvement:.2%}",
\n "confidence_level": f"{(1 – best_p_value):.2%}",
\n "next_steps": [
\n "Roll out to 100% of traffic",
\n "Monitor for any negative side effects",
\n "Update documentation"
\n ]
\n }
\n else:
\n # \u68c0\u67e5\u662f\u5426\u6709\u663e\u8457\u53d8\u5dee\u7684\u5206\u7ec4
\n worst_group = None
\n worst_deterioration = 0<\/p>\n

for treatment_group, comparison in comparisons.items():
\n event_comparisons = comparison.get('event_comparisons', {})<\/p>\n

for event_type, event_comp in event_comparisons.items():
\n if event_comp.get('significant', False):
\n improvement = event_comp.get('relative_improvement', 0)
\n if improvement < worst_deterioration:
\n worst_deterioration = improvement
\n worst_group = treatment_group<\/p>\n

if worst_group and worst_deterioration < -0.05: # \u5982\u679c\u53d8\u5dee\u8d85\u8fc75%
\n return {
\n "decision": "reject",
\n "rejected_group": worst_group,
\n "deterioration": f"{abs(worst_deterioration):.2%}",
\n "reason": f"Group {worst_group} performed significantly worse than control",
\n "next_steps": [
\n "Stop traffic to this group",
\n "Analyze why performance was worse",
\n "Consider alternative approaches"
\n ]
\n }
\n else:
\n return {
\n "decision": "continue_testing",
\n "reason": "No significant difference detected",
\n "next_steps": [
\n "Increase sample size if possible",
\n "Extend testing duration",
\n "Consider adjusting experiment parameters"
\n ]
\n }<\/p>\n

# ==================== \u5b9e\u65f6\u76d1\u63a7\u5668 ====================
\nclass ExperimentMonitor:
\n """\u5b9e\u9a8c\u5b9e\u65f6\u76d1\u63a7\u5668"""<\/p>\n

def __init__(self, ab_test_manager: ABTestManager, analysis: ExperimentAnalysis):
\n self.ab_test_manager = ab_test_manager
\n self.analysis = analysis
\n self.monitoring_tasks = {}
\n self.alert_rules = {}
\n self.logger = logging.getLogger("ExperimentMonitor")<\/p>\n

# \u9ed8\u8ba4\u62a5\u8b66\u89c4\u5219
\n self.default_alert_rules = {
\n 'sample_size': {
\n 'min_users': 100,
\n 'warning_threshold': 50
\n },
\n 'conversion_rate': {
\n 'min_difference': 0.05, # 5%\u5dee\u5f02
\n 'significance_level': 0.05
\n },
\n 'safety_metrics': {
\n 'max_deterioration': -0.1, # \u6700\u5927\u5141\u8bb8\u7684\u4e0b\u964d
\n 'check_frequency': 'hourly'
\n }
\n }<\/p>\n

async def start_monitoring(self, experiment_name: str):
\n """\u5f00\u59cb\u76d1\u63a7\u5b9e\u9a8c"""
\n if experiment_name in self.monitoring_tasks:
\n self.logger.warning(f"Experiment {experiment_name} is already being monitored")
\n return<\/p>\n

# \u521b\u5efa\u76d1\u63a7\u4efb\u52a1
\n task = asyncio.create_task(self._monitor_experiment(experiment_name))
\n self.monitoring_tasks[experiment_name] = task<\/p>\n

self.logger.info(f"Started monitoring experiment: {experiment_name}")<\/p>\n

async def stop_monitoring(self, experiment_name: str):
\n """\u505c\u6b62\u76d1\u63a7\u5b9e\u9a8c"""
\n if experiment_name not in self.monitoring_tasks:
\n return<\/p>\n

task = self.monitoring_tasks[experiment_name]
\n task.cancel()<\/p>\n

try:
\n await task
\n except asyncio.CancelledError:
\n pass<\/p>\n

del self.monitoring_tasks[experiment_name]
\n self.logger.info(f"Stopped monitoring experiment: {experiment_name}")<\/p>\n

async def _monitor_experiment(self, experiment_name: str):
\n """\u76d1\u63a7\u5b9e\u9a8c\u4e3b\u5faa\u73af"""
\n try:
\n while True:
\n # \u5206\u6790\u5b9e\u9a8c\u6570\u636e
\n analysis_result = await self.analysis.analyze_experiment(
\n experiment_name,
\n start_date=datetime.now() – timedelta(hours=24) # \u6700\u8fd124\u5c0f\u65f6
\n )<\/p>\n

if 'error' not in analysis_result:
\n # \u68c0\u67e5\u62a5\u8b66\u6761\u4ef6
\n alerts = await self._check_alerts(analysis_result)<\/p>\n

if alerts:
\n await self._send_alerts(experiment_name, alerts)<\/p>\n

# \u68c0\u67e5\u662f\u5426\u8fbe\u5230\u505c\u6b62\u6761\u4ef6
\n stop_recommended = await self._check_stop_conditions(analysis_result)<\/p>\n

if stop_recommended:
\n self.logger.info(f"Stop condition met for experiment: {experiment_name}")
\n await self.ab_test_manager.stop_experiment(experiment_name)
\n break<\/p>\n

# \u7b49\u5f85\u4e00\u6bb5\u65f6\u95f4\u540e\u518d\u6b21\u68c0\u67e5
\n await asyncio.sleep(3600) # \u6bcf\u5c0f\u65f6\u68c0\u67e5\u4e00\u6b21<\/p>\n

except asyncio.CancelledError:
\n raise
\n except Exception as e:
\n self.logger.error(f"Error monitoring experiment {experiment_name}: {e}")<\/p>\n

async def _check_alerts(self, analysis_result: Dict[str, Any]) -> List[Dict[str, Any]]:
\n """\u68c0\u67e5\u62a5\u8b66\u6761\u4ef6"""
\n alerts = []<\/p>\n

# \u68c0\u67e5\u6837\u672c\u91cf
\n summary = analysis_result.get('summary', {})
\n for group_name, group_info in summary.get('groups', {}).items():
\n user_count = group_info.get('user_count', 0)<\/p>\n

if user_count < self.default_alert_rules['sample_size']['warning_threshold']:
\n alerts.append({
\n 'type': 'warning',
\n 'code': 'LOW_SAMPLE_SIZE',
\n 'group': group_name,
\n 'message': f'Group {group_name} has only {user_count} users',
\n 'severity': 'low'
\n })
\n elif user_count < self.default_alert_rules['sample_size']['min_users']:
\n alerts.append({
\n 'type': 'warning',
\n 'code': 'MIN_SAMPLE_NOT_REACHED',
\n 'group': group_name,
\n 'message': f'Group {group_name} has not reached minimum sample size',
\n 'severity': 'medium'
\n })<\/p>\n

# \u68c0\u67e5\u5b89\u5168\u6027\u6307\u6807
\n group_comparison = analysis_result.get('group_comparison', {})
\n comparisons = group_comparison.get('comparisons', {})<\/p>\n

for treatment_group, comparison in comparisons.items():
\n event_comparisons = comparison.get('event_comparisons', {})<\/p>\n

for event_type, event_comp in event_comparisons.items():
\n relative_improvement = event_comp.get('relative_improvement', 0)<\/p>\n

if relative_improvement < self.default_alert_rules['safety_metrics']['max_deterioration']:
\n alerts.append({
\n 'type': 'critical',
\n 'code': 'SAFETY_THRESHOLD_BREACHED',
\n 'group': treatment_group,
\n 'event_type': event_type,
\n 'message': f'Group {treatment_group} shows {relative_improvement:.2%} deterioration in {event_type}',
\n 'severity': 'high'
\n })<\/p>\n

return alerts<\/p>\n

async def _send_alerts(self, experiment_name: str, alerts: List[Dict[str, Any]]):
\n """\u53d1\u9001\u62a5\u8b66"""
\n for alert in alerts:
\n # \u8fd9\u91cc\u53ef\u4ee5\u5b9e\u73b0\u62a5\u8b66\u53d1\u9001\u903b\u8f91
\n # \u4f8b\u5982：\u53d1\u9001\u5230Slack\u3001\u90ae\u4ef6\u3001\u77ed\u4fe1\u7b49
\n alert_message = (
\n f"🚨 Experiment Alert: {experiment_name}\\\\n"
\n f"Type: {alert['type']}\\\\n"
\n f"Code: {alert['code']}\\\\n"
\n f"Message: {alert['message']}\\\\n"
\n f"Severity: {alert['severity']}"
\n )<\/p>\n

self.logger.warning(alert_message)<\/p>\n

# \u53ef\u4ee5\u6839\u636e\u4e25\u91cd\u7a0b\u5ea6\u91c7\u53d6\u4e0d\u540c\u884c\u52a8
\n if alert['severity'] == 'high':
\n # \u7d27\u6025\u62a5\u8b66：\u53ef\u80fd\u9700\u8981\u7acb\u5373\u505c\u6b62\u5b9e\u9a8c
\n await self._send_urgent_alert(alert_message)
\n elif alert['severity'] == 'medium':
\n # \u4e2d\u7b49\u62a5\u8b66：\u53d1\u9001\u5230\u76d1\u63a7\u9891\u9053
\n await self._send_monitoring_alert(alert_message)
\n else:
\n # \u4f4e\u7ea7\u522b\u62a5\u8b66：\u8bb0\u5f55\u65e5\u5fd7
\n self.logger.info(f"Low severity alert: {alert_message}")<\/p>\n

async def _send_urgent_alert(self, message: str):
\n """\u53d1\u9001\u7d27\u6025\u62a5\u8b66"""
\n # \u5b9e\u73b0\u7d27\u6025\u62a5\u8b66\u903b\u8f91
\n pass<\/p>\n

async def _send_monitoring_alert(self, message: str):
\n """\u53d1\u9001\u76d1\u63a7\u62a5\u8b66"""
\n # \u5b9e\u73b0\u76d1\u63a7\u62a5\u8b66\u903b\u8f91
\n pass<\/p>\n

async def _check_stop_conditions(self, analysis_result: Dict[str, Any]) -> bool:
\n """\u68c0\u67e5\u505c\u6b62\u6761\u4ef6"""
\n group_comparison = analysis_result.get('group_comparison', {})
\n recommendation = group_comparison.get('overall_recommendation', {})<\/p>\n

decision = recommendation.get('decision', '')<\/p>\n

# \u5982\u679c\u63a8\u8350\u5b9e\u65bd\u6216\u62d2\u7edd，\u5efa\u8bae\u505c\u6b62\u5b9e\u9a8c
\n if decision in ['implement', 'reject']:
\n return True<\/p>\n

# \u68c0\u67e5\u662f\u5426\u8fbe\u5230\u6700\u5c0f\u6837\u672c\u91cf
\n summary = analysis_result.get('summary', {})
\n min_users_met = True<\/p>\n

for group_info in summary.get('groups', {}).values():
\n if group_info.get('user_count', 0) < self.default_alert_rules['sample_size']['min_users']:
\n min_users_met = False
\n break<\/p>\n

# \u5982\u679c\u5df2\u7ecf\u8fbe\u5230\u6700\u5c0f\u6837\u672c\u91cf\u4f46\u4ecd\u7136\u6ca1\u6709\u663e\u8457\u5dee\u5f02，\u53ef\u4ee5\u8003\u8651\u505c\u6b62
\n if min_users_met and decision == 'continue_testing':
\n # \u53ef\u4ee5\u6dfb\u52a0\u66f4\u591a\u505c\u6b62\u6761\u4ef6
\n pass<\/p>\n

return False<\/p>\n

def set_alert_rules(self, experiment_name: str, rules: Dict[str, Any]):
\n """\u8bbe\u7f6e\u62a5\u8b66\u89c4\u5219"""
\n self.alert_rules[experiment_name] = rules<\/p>\n

def get_monitoring_status(self) -> Dict[str, Any]:
\n """\u83b7\u53d6\u76d1\u63a7\u72b6\u6001"""
\n status = {
\n 'timestamp': datetime.now().isoformat(),
\n 'monitoring_experiments': list(self.monitoring_tasks.keys()),
\n 'total_monitored': len(self.monitoring_tasks)
\n }<\/p>\n

return status<\/p>\n

# ==================== \u81ea\u52a8\u5316\u62a5\u544a\u7cfb\u7edf ====================
\nclass AutomatedReporting:
\n """\u81ea\u52a8\u5316\u62a5\u544a\u7cfb\u7edf"""<\/p>\n

def __init__(self, analysis: ExperimentAnalysis):
\n self.analysis = analysis
\n self.report_templates = self._load_report_templates()
\n self.scheduled_reports = {}<\/p>\n

def _load_report_templates(self) -> Dict[str, str]:
\n """\u52a0\u8f7d\u62a5\u544a\u6a21\u677f"""
\n templates = {
\n 'executive': """
\n # Experiment Report: {experiment_name}<\/p>\n

## Executive Summary
\n {executive_summary}<\/p>\n

## Key Findings
\n {key_findings}<\/p>\n

## Recommendations
\n {recommendations}<\/p>\n

## Next Steps
\n {next_steps}
\n """,<\/p>\n

'technical': """
\n # Technical Analysis Report: {experiment_name}<\/p>\n

## Methodology
\n {methodology}<\/p>\n

## Statistical Analysis
\n {statistical_analysis}<\/p>\n

## Results
\n {results}<\/p>\n

## Appendix: Detailed Metrics
\n {detailed_metrics}
\n """,<\/p>\n

'dashboard': """
\n # Experiment Dashboard: {experiment_name}<\/p>\n

## Overview
\n {overview}<\/p>\n

## Real-time Metrics
\n {realtime_metrics}<\/p>\n

## Performance Charts
\n {performance_charts}<\/p>\n

## Alerts
\n {alerts}
\n """
\n }<\/p>\n

return templates<\/p>\n

async def generate_report(
\n self,
\n experiment_name: str,
\n report_type: str = 'executive',
\n start_date: Optional[datetime] = None,
\n end_date: Optional[datetime] = None
\n ) -> Dict[str, Any]:
\n """\u751f\u6210\u62a5\u544a"""
\n if report_type not in self.report_templates:
\n raise ValueError(f"Unknown report type: {report_type}")<\/p>\n

# \u5206\u6790\u5b9e\u9a8c\u6570\u636e
\n analysis_result = await self.analysis.analyze_experiment(
\n experiment_name, start_date, end_date
\n )<\/p>\n

if 'error' in analysis_result:
\n return analysis_result<\/p>\n

# \u6839\u636e\u62a5\u544a\u7c7b\u578b\u683c\u5f0f\u5316\u6570\u636e
\n formatted_data = await self._format_report_data(
\n analysis_result, report_type
\n )<\/p>\n

# \u586b\u5145\u6a21\u677f
\n template = self.report_templates[report_type]
\n report_content = template.format(**formatted_data)<\/p>\n

# \u6dfb\u52a0\u53ef\u89c6\u5316\u56fe\u8868
\n if 'visualizations' in analysis_result:
\n report_content += "\\\\n\\\\n## Visualizations\\\\n"
\n for viz_name, viz_html in analysis_result['visualizations'].items():
\n if viz_name != 'error':
\n report_content += f"\\\\n### {viz_name.replace('_', ' ').title()}\\\\n"
\n report_content += f"{viz_html}\\\\n"<\/p>\n

report = {
\n 'experiment_name': experiment_name,
\n 'report_type': report_type,
\n 'generated_at': datetime.now().isoformat(),
\n 'time_period': {
\n 'start': start_date.isoformat() if start_date else analysis_result['time_period']['start'],
\n 'end': end_date.isoformat() if end_date else analysis_result['time_period']['end']
\n },
\n 'content': report_content,
\n 'analysis_data': analysis_result
\n }<\/p>\n

return report<\/p>\n

async def _format_report_data(
\n self,
\n analysis_result: Dict[str, Any],
\n report_type: str
\n ) -> Dict[str, str]:
\n """\u683c\u5f0f\u5316\u62a5\u544a\u6570\u636e"""
\n formatted_data = {
\n 'experiment_name': analysis_result['experiment_name'],
\n 'executive_summary': self._generate_executive_summary(analysis_result),
\n 'key_findings': self._generate_key_findings(analysis_result),
\n 'recommendations': analysis_result.get('group_comparison', {}).get('overall_recommendation', {}).get('next_steps', ['No recommendations']),
\n 'next_steps': self._generate_next_steps(analysis_result),
\n 'methodology': self._format_methodology(analysis_result),
\n 'statistical_analysis': self._format_statistical_analysis(analysis_result),
\n 'results': self._format_results(analysis_result),
\n 'detailed_metrics': self._format_detailed_metrics(analysis_result),
\n 'overview': self._format_overview(analysis_result),
\n 'realtime_metrics': self._format_realtime_metrics(analysis_result),
\n 'performance_charts': "Performance charts will be embedded here",
\n 'alerts': "No active alerts"
\n }<\/p>\n

return formatted_data<\/p>\n

def _generate_executive_summary(self, analysis_result: Dict[str, Any]) -> str:
\n """\u751f\u6210\u6267\u884c\u6458\u8981"""
\n summary = analysis_result.get('summary', {})
\n comparison = analysis_result.get('group_comparison', {})<\/p>\n

total_users = summary.get('total_users', 0)
\n recommendation = comparison.get('overall_recommendation', {})<\/p>\n

exec_summary = f"""
\n This experiment involved {total_users} users over the analysis period.<\/p>\n

Key outcome: {recommendation.get('decision', 'No decision reached')}.<\/p>\n

Primary metric performance varied across groups, with treatment groups showing
\n {'improvement' if recommendation.get('decision') == 'implement' else 'no significant change'}
\n compared to the control group.
\n """<\/p>\n

return exec_summary<\/p>\n

def _generate_key_findings(self, analysis_result: Dict[str, Any]) -> str:
\n """\u751f\u6210\u5173\u952e\u53d1\u73b0"""
\n comparison = analysis_result.get('group_comparison', {})
\n comparisons = comparison.get('comparisons', {})<\/p>\n

findings = []<\/p>\n

for treatment_group, comp_data in comparisons.items():
\n event_comparisons = comp_data.get('event_comparisons', {})<\/p>\n

for event_type, event_comp in event_comparisons.items():
\n if event_comp.get('significant', False):
\n improvement = event_comp.get('relative_improvement', 0)
\n p_value = event_comp.get('p_value', 1.0)<\/p>\n

findings.append(
\n f"\u2022 Group {treatment_group} showed {improvement:.2%} improvement "
\n f"in {event_type} (p={p_value:.4f})"
\n )<\/p>\n

if not findings:
\n findings.append("\u2022 No statistically significant differences were found between groups")<\/p>\n

return "\\\\n".join(findings)<\/p>\n

def _generate_next_steps(self, analysis_result: Dict[str, Any]) -> str:
\n """\u751f\u6210\u540e\u7eed\u6b65\u9aa4"""
\n recommendation = analysis_result.get('group_comparison', {}).get('overall_recommendation', {})
\n decision = recommendation.get('decision', '')<\/p>\n

if decision == 'implement':
\n return """
\n 1. Roll out the winning variant to 100% of traffic
\n 2. Monitor key metrics for any negative impact
\n 3. Document the change and update relevant systems
\n 4. Plan follow-up experiments for further optimization
\n """
\n elif decision == 'reject':
\n return """
\n 1. Stop traffic to the underperforming variant
\n 2. Analyze reasons for poor performance
\n 3. Consider alternative approaches
\n 4. Document learnings for future experiments
\n """
\n else:
\n return """
\n 1. Continue running the experiment
\n 2. Increase sample size if possible
\n 3. Consider adjusting experiment parameters
\n 4. Set up additional monitoring
\n """<\/p>\n

def _format_methodology(self, analysis_result: Dict[str, Any]) -> str:
\n """\u683c\u5f0f\u5316\u65b9\u6cd5\u8bba"""
\n methodology = """
\n ## Experimental Design
\n – Random assignment of users to treatment groups
\n – Control group vs. treatment group(s) comparison
\n – Minimum sample size: 100 users per group<\/p>\n

## Statistical Methods
\n – Conversion rates compared using proportion z-tests
\n – Continuous metrics compared using t-tests
\n – Significance level: \u03b1 = 0.05
\n – Confidence intervals: 95%<\/p>\n

## Data Collection
\n – Real-time event tracking
\n – User-level attribution
\n – Time-series analysis
\n """<\/p>\n

return methodology<\/p>\n

def _format_statistical_analysis(self, analysis_result: Dict[str, Any]) -> str:
\n """\u683c\u5f0f\u5316\u7edf\u8ba1\u5206\u6790"""
\n power_analysis = analysis_result.get('power_analysis', {})
\n comparison = analysis_result.get('group_comparison', {})<\/p>\n

stats_text = "## Statistical Power Analysis\\\\n\\\\n"<\/p>\n

for group_name, power_info in power_analysis.items():
\n if 'error' not in power_info:
\n stats_text += (
\n f"**{group_name}**:\\\\n"
\n f"- Current sample: {power_info.get('current_sample_size', 0)}\\\\n"
\n f"- Required sample: {power_info.get('required_sample_size', 'N\/A')}\\\\n"
\n f"- Sufficient power: {power_info.get('sufficient_power', False)}\\\\n\\\\n"
\n )<\/p>\n

stats_text += "## Significance Testing Results\\\\n\\\\n"<\/p>\n

comparisons = comparison.get('comparisons', {})
\n for treatment_group, comp_data in comparisons.items():
\n stats_text += f"### {treatment_group} vs Control\\\\n"<\/p>\n

event_comparisons = comp_data.get('event_comparisons', {})
\n for event_type, event_comp in event_comparisons.items():
\n stats_text += (
\n f"**{event_type}**: "
\n f"p-value = {event_comp.get('p_value', 1.0):.4f}, "
\n f"Significant: {event_comp.get('significant', False)}\\\\n"
\n )<\/p>\n

return stats_text<\/p>\n

def _format_results(self, analysis_result: Dict[str, Any]) -> str:
\n """\u683c\u5f0f\u5316\u7ed3\u679c"""
\n summary = analysis_result.get('summary', {})<\/p>\n

results_text = "## Overall Results\\\\n\\\\n"
\n results_text += f"Total Users: {summary.get('total_users', 0)}\\\\n"
\n results_text += f"Total Events: {summary.get('total_events', 0)}\\\\n"
\n results_text += f"Events per User: {summary.get('events_per_user', 0):.2f}\\\\n\\\\n"<\/p>\n

results_text += "## Group Performance\\\\n\\\\n"<\/p>\n

for group_name, group_info in summary.get('groups', {}).items():
\n results_text += f"### {group_name}\\\\n"
\n results_text += f"- Users: {group_info.get('user_count', 0)}\\\\n"
\n results_text += f"- Events: {group_info.get('event_count', 0)}\\\\n"<\/p>\n

# \u8f6c\u5316\u7387
\n event_dist = group_info.get('event_distribution', {})
\n user_count = group_info.get('user_count', 1)<\/p>\n

for event_type, count in event_dist.items():
\n rate = count \/ user_count if user_count > 0 else 0
\n results_text += f"- {event_type}: {count} events ({rate:.2%})\\\\n"<\/p>\n

results_text += "\\\\n"<\/p>\n

return results_text<\/p>\n

def _format_detailed_metrics(self, analysis_result: Dict[str, Any]) -> str:
\n """\u683c\u5f0f\u5316\u8be6\u7ec6\u6307\u6807"""
\n summary = analysis_result.get('summary', {})<\/p>\n

metrics_text = "## Detailed Metrics by Group\\\\n\\\\n"<\/p>\n

for group_name, group_info in summary.get('groups', {}).items():
\n metrics_text += f"### {group_name}\\\\n"<\/p>\n

event_stats = group_info.get('event_statistics', {})
\n if event_stats:
\n for event_type, stats in event_stats.items():
\n metrics_text += (
\n f"**{event_type}**:\\\\n"
\n f"- Count: {stats.get('count', 0)}\\\\n"
\n f"- Mean: {stats.get('mean', 0):.2f}\\\\n"
\n f"- Std: {stats.get('std', 0):.2f}\\\\n"
\n f"- Min: {stats.get('min', 0):.2f}\\\\n"
\n f"- Max: {stats.get('max', 0):.2f}\\\\n"
\n f"- Median: {stats.get('median', 0):.2f}\\\\n\\\\n"
\n )
\n else:
\n metrics_text += "No numeric event data available\\\\n\\\\n"<\/p>\n

return metrics_text<\/p>\n

def _format_overview(self, analysis_result: Dict[str, Any]) -> str:
\n """\u683c\u5f0f\u5316\u6982\u89c8"""
\n return """
\n ## Experiment Status
\n – Status: Running
\n – Start Date: {start_date}
\n – Duration: {duration_days} days
\n – Sample Rate: 10%<\/p>\n

## Key Metrics
\n – Primary Metric: Conversion Rate
\n – Guardrail Metrics: Revenue, User Satisfaction
\n – Statistical Power: 80%<\/p>\n

## Current Allocation
\n – Control: 50%
\n – Treatment A: 25%
\n – Treatment B: 25%
\n """.format(
\n start_date=analysis_result.get('time_period', {}).get('start', 'N\/A'),
\n duration_days=(datetime.now() – datetime.fromisoformat(
\n analysis_result.get('time_period', {}).get('start', datetime.now().isoformat())
\n )).days
\n )<\/p>\n

def _format_realtime_metrics(self, analysis_result: Dict[str, Any]) -> str:
\n """\u683c\u5f0f\u5316\u5b9e\u65f6\u6307\u6807"""
\n time_series = analysis_result.get('time_series_analysis', {}).get('time_series_data', {})<\/p>\n

if not time_series:
\n return "No time series data available"<\/p>\n

latest_data = {}
\n for group_name, group_data in time_series.items():
\n if group_data['dates'] and group_data['daily_conversion_rates']:
\n latest_idx = -1
\n latest_data[group_name] = {
\n 'date': group_data['dates'][latest_idx],
\n 'conversion_rate': group_data['daily_conversion_rates'][latest_idx],
\n 'users': group_data['daily_users'][latest_idx]
\n }<\/p>\n

metrics_text = "## Latest Metrics (Last 24 Hours)\\\\n\\\\n"<\/p>\n

for group_name, data in latest_data.items():
\n metrics_text += (
\n f"**{group_name}**:\\\\n"
\n f"- Date: {data['date']}\\\\n"
\n f"- Conversion Rate: {data['conversion_rate']:.2%}\\\\n"
\n f"- Users: {data['users']}\\\\n\\\\n"
\n )<\/p>\n

return metrics_text<\/p>\n

def schedule_report(
\n self,
\n experiment_name: str,
\n frequency: str,
\n recipients: List[str]
\n ):
\n """\u5b89\u6392\u5b9a\u671f\u62a5\u544a"""
\n schedule_id = f"{experiment_name}_{frequency}_{datetime.now().timestamp()}"<\/p>\n

self.scheduled_reports[schedule_id] = {
\n 'experiment_name': experiment_name,
\n 'frequency': frequency,
\n 'recipients': recipients,
\n 'last_sent': None,
\n 'next_scheduled': self._calculate_next_run(frequency)
\n }<\/p>\n

return schedule_id<\/p>\n

def _calculate_next_run(self, frequency: str) -> datetime:
\n """\u8ba1\u7b97\u4e0b\u6b21\u8fd0\u884c\u65f6\u95f4"""
\n now = datetime.now()<\/p>\n

if frequency == 'daily':
\n return now + timedelta(days=1)
\n elif frequency == 'weekly':
\n return now + timedelta(weeks=1)
\n elif frequency == 'monthly':
\n # \u7b80\u5316\u5904\u7406：\u4e0b\u4e2a\u6708\u7684\u540c\u4e00\u5929
\n if now.month == 12:
\n return datetime(now.year + 1, 1, now.day)
\n else:
\n return datetime(now.year, now.month + 1, now.day)
\n else:
\n return now + timedelta(days=1) # \u9ed8\u8ba4\u6bcf\u5929<\/p>\n

async def send_scheduled_reports(self):
\n """\u53d1\u9001\u5b9a\u671f\u62a5\u544a"""
\n now = datetime.now()<\/p>\n

for schedule_id, schedule in self.scheduled_reports.items():
\n next_run = schedule['next_scheduled']<\/p>\n

if now >= next_run:
\n # \u751f\u6210\u5e76\u53d1\u9001\u62a5\u544a
\n report = await self.generate_report(
\n schedule['experiment_name'],
\n report_type='executive'
\n )<\/p>\n

# \u53d1\u9001\u62a5\u544a（\u8fd9\u91cc\u9700\u8981\u5b9e\u73b0\u53d1\u9001\u903b\u8f91）
\n await self._send_report(report, schedule['recipients'])<\/p>\n

# \u66f4\u65b0\u8ba1\u5212
\n schedule['last_sent'] = now
\n schedule['next_scheduled'] = self._calculate_next_run(schedule['frequency'])<\/p>\n

async def _send_report(self, report: Dict[str, Any], recipients: List[str]):
\n """\u53d1\u9001\u62a5\u544a"""
\n # \u8fd9\u91cc\u53ef\u4ee5\u5b9e\u73b0\u90ae\u4ef6\u53d1\u9001\u3001Slack\u901a\u77e5\u7b49\u903b\u8f91
\n print(f"Would send report to: {recipients}")
\n print(f"Report content length: {len(str(report))} characters")<\/p>\n

# ==================== \u4e3b\u7a0b\u5e8f\u5165\u53e3 ====================
\nasync def main():
\n """\u4e3b\u7a0b\u5e8f"""
\n # \u521d\u59cb\u5316\u65e5\u5fd7
\n logging.basicConfig(
\n level=logging.INFO,
\n format='%(asctime)s – %(name)s – %(levelname)s – %(message)s'
\n )<\/p>\n

logger = logging.getLogger(__name__)<\/p>\n

try:
\n # 1. \u521b\u5efaA\/B\u6d4b\u8bd5\u7ba1\u7406\u5668
\n config = create_sample_config()
\n ab_test_manager = ABTestManager(config)<\/p>\n

# 2. \u521b\u5efa\u5b9e\u9a8c\u5206\u6790\u5668
\n experiment_analysis = ExperimentAnalysis()<\/p>\n

# 3. \u521b\u5efa\u5b9e\u9a8c\u76d1\u63a7\u5668
\n experiment_monitor = ExperimentMonitor(ab_test_manager, experiment_analysis)<\/p>\n

# 4. \u521b\u5efa\u81ea\u52a8\u5316\u62a5\u544a\u7cfb\u7edf
\n reporting = AutomatedReporting(experiment_analysis)<\/p>\n

# 5. \u6a21\u62df\u4e00\u4e9b\u6570\u636e
\n logger.info("Simulating experiment data…")
\n for i in range(1000):
\n user_id = f"user_{i}"<\/p>\n

# \u5206\u914d\u5b9e\u9a8c
\n assignment = ab_test_manager.assign_experiment(
\n user_id=user_id,
\n experiment_name='recommendation_model_v2'
\n )<\/p>\n

if assignment:
\n # \u8bb0\u5f55\u5206\u914d\u4e8b\u4ef6
\n experiment_analysis.record_event({
\n 'experiment_name': 'recommendation_model_v2',
\n 'user_id': user_id,
\n 'group_name': assignment['group_name'],
\n 'event_type': 'assignment',
\n 'timestamp': datetime.now(),
\n 'metadata': {'assignment_id': assignment['assignment_id']}
\n })<\/p>\n

# \u6a21\u62df\u7528\u6237\u884c\u4e3a（\u63a7\u5236\u7ec4\u548c\u5b9e\u9a8c\u7ec4\u8868\u73b0\u4e0d\u540c）
\n if assignment['group_name'] == 'control':
\n click_prob = 0.08 # \u63a7\u5236\u7ec4\u70b9\u51fb\u73878%
\n purchase_prob = 0.15 # \u70b9\u51fb\u540e\u7684\u8d2d\u4e70\u738715%
\n elif assignment['group_name'] == 'treatment_v2':
\n click_prob = 0.10 # \u5b9e\u9a8c\u7ec4v2\u70b9\u51fb\u738710%
\n purchase_prob = 0.18 # \u70b9\u51fb\u540e\u7684\u8d2d\u4e70\u738718%
\n else: # treatment_v3
\n click_prob = 0.12 # \u5b9e\u9a8c\u7ec4v3\u70b9\u51fb\u738712%
\n purchase_prob = 0.20 # \u70b9\u51fb\u540e\u7684\u8d2d\u4e70\u738720%<\/p>\n

# \u6a21\u62df\u70b9\u51fb\u4e8b\u4ef6
\n if random.random() < click_prob:
\n experiment_analysis.record_event({
\n 'experiment_name': 'recommendation_model_v2',
\n 'user_id': user_id,
\n 'group_name': assignment['group_name'],
\n 'event_type': 'click',
\n 'timestamp': datetime.now(),
\n 'metadata': {'source': 'recommendation'}
\n })<\/p>\n

# \u6a21\u62df\u8d2d\u4e70\u4e8b\u4ef6
\n if random.random() < purchase_prob:
\n purchase_amount = random.uniform(10, 100)
\n experiment_analysis.record_event({
\n 'experiment_name': 'recommendation_model_v2',
\n 'user_id': user_id,
\n 'group_name': assignment['group_name'],
\n 'event_type': 'purchase',
\n 'event_value': purchase_amount,
\n 'timestamp': datetime.now(),
\n 'metadata': {
\n 'amount': purchase_amount,
\n 'items': random.randint(1, 5)
\n }
\n })<\/p>\n

# 6. \u5206\u6790\u5b9e\u9a8c\u6570\u636e
\n logger.info("Analyzing experiment data…")
\n analysis_result = await experiment_analysis.analyze_experiment(
\n 'recommendation_model_v2'
\n )<\/p>\n

# 7. \u751f\u6210\u62a5\u544a
\n logger.info("Generating report…")
\n report = await reporting.generate_report(
\n 'recommendation_model_v2',
\n report_type='executive'
\n )<\/p>\n

# 8. \u8f93\u51fa\u7ed3\u679c
\n print("\\\\n" + "="*80)
\n print("EXPERIMENT ANALYSIS REPORT")
\n print("="*80)
\n print(f"Experiment: {report['experiment_name']}")
\n print(f"Report Type: {report['report_type']}")
\n print(f"Generated: {report['generated_at']}")
\n print("\\\\n" + "="*80)
\n print("CONTENT")
\n print("="*80)
\n print(report['content'][:2000] + "…" if len(report['content']) > 2000 else report['content'])<\/p>\n

# 9. \u5f00\u59cb\u76d1\u63a7
\n logger.info("Starting experiment monitoring…")
\n await experiment_monitor.start_monitoring('recommendation_model_v2')<\/p>\n

# \u7b49\u5f85\u4e00\u6bb5\u65f6\u95f4（\u6a21\u62df\u76d1\u63a7\u8fd0\u884c）
\n await asyncio.sleep(5)<\/p>\n

# 10. \u68c0\u67e5\u76d1\u63a7\u72b6\u6001
\n status = experiment_monitor.get_monitoring_status()
\n print(f"\\\\nMonitoring Status: {status}")<\/p>\n

except Exception as e:
\n logger.error(f"Error in main: {e}")
\n raise<\/p>\n

if __name__ == "__main__":
\n asyncio.run(main()) <\/p>\n

\u4e09\u3001\u751f\u4ea7\u73af\u5883\u90e8\u7f72\u4e0e\u4f18\u5316<\/h3>\n

3.1 \u5bb9\u5668\u5316\u90e8\u7f72\u914d\u7f6e<\/h4>\n

yaml<\/p>\n<\/p>\n

\u590d\u5236<\/p>\n<\/p>\n

\u4e0b\u8f7d<\/p>\n

# docker-compose.yaml
\nversion: '3.8'<\/p>\n

services:
\n # \u6a21\u578b\u670d\u52a1API
\n ml-model-api:
\n build:
\n context: .
\n dockerfile: Dockerfile.api
\n ports:
\n – "8000:8000"
\n environment:
\n – ENVIRONMENT=production
\n – LOG_LEVEL=INFO
\n – MODEL_CONFIG_PATH=\/app\/config\/models.json
\n – AB_TEST_CONFIG_PATH=\/app\/config\/ab_test.json
\n – DATABASE_URL=postgresql:\/\/user:password@db:5432\/ml_models
\n volumes:
\n – .\/models:\/app\/models
\n – .\/config:\/app\/config
\n – .\/logs:\/app\/logs
\n depends_on:
\n – db
\n – redis
\n – prometheus
\n networks:
\n – ml-network
\n healthcheck:
\n test: ["CMD", "curl", "-f", "http:\/\/localhost:8000\/health"]
\n interval: 30s
\n timeout: 10s
\n retries: 3
\n deploy:
\n replicas: 3
\n resources:
\n limits:
\n cpus: '2'
\n memory: 4G
\n reservations:
\n cpus: '0.5'
\n memory: 1G<\/p>\n

# \u7279\u5f81\u5de5\u7a0b\u670d\u52a1
\n feature-service:
\n build:
\n context: .
\n dockerfile: Dockerfile.feature
\n environment:
\n – REDIS_HOST=redis
\n – FEATURE_STORE_TYPE=redis
\n – LOG_LEVEL=INFO
\n volumes:
\n – .\/features:\/app\/features
\n depends_on:
\n – redis
\n networks:
\n – ml-network
\n deploy:
\n replicas: 2<\/p>\n

# A\/B\u6d4b\u8bd5\u670d\u52a1
\n ab-test-service:
\n build:
\n context: .
\n dockerfile: Dockerfile.abtest
\n ports:
\n – "8001:8001"
\n environment:
\n – DATABASE_URL=postgresql:\/\/user:password@db:5432\/ab_test
\n – REDIS_HOST=redis
\n – CACHE_TTL=3600
\n depends_on:
\n – db
\n – redis
\n networks:
\n – ml-network
\n healthcheck:
\n test: ["CMD", "curl", "-f", "http:\/\/localhost:8001\/health"]<\/p>\n

# \u6570\u636e\u5e93 (PostgreSQL)
\n db:
\n image: postgres:14
\n environment:
\n – POSTGRES_USER=user
\n – POSTGRES_PASSWORD=password
\n – POSTGRES_DB=ml_models
\n volumes:
\n – postgres_data:\/var\/lib\/postgresql\/data
\n – .\/init.sql:\/docker-entrypoint-initdb.d\/init.sql
\n networks:
\n – ml-network
\n healthcheck:
\n test: ["CMD-SHELL", "pg_isready -U user"]
\n interval: 10s
\n timeout: 5s
\n retries: 5<\/p>\n

# Redis\u7f13\u5b58
\n redis:
\n image: redis:7-alpine
\n command: redis-server –appendonly yes
\n volumes:
\n – redis_data:\/data
\n networks:
\n – ml-network
\n healthcheck:
\n test: ["CMD", "redis-cli", "ping"]
\n interval: 10s
\n timeout: 5s
\n retries: 3<\/p>\n

# \u76d1\u63a7 (Prometheus + Grafana)
\n prometheus:
\n image: prom\/prometheus
\n volumes:
\n – .\/prometheus.yml:\/etc\/prometheus\/prometheus.yml
\n – prometheus_data:\/prometheus
\n ports:
\n – "9090:9090"
\n networks:
\n – ml-network<\/p>\n

grafana:
\n image: grafana\/grafana
\n environment:
\n – GF_SECURITY_ADMIN_PASSWORD=admin
\n volumes:
\n – grafana_data:\/var\/lib\/grafana
\n – .\/grafana\/dashboards:\/etc\/grafana\/provisioning\/dashboards
\n ports:
\n – "3000:3000"
\n networks:
\n – ml-network
\n depends_on:
\n – prometheus<\/p>\n

# \u6a21\u578b\u8bad\u7ec3\u6d41\u6c34\u7ebf (Airflow)
\n airflow-scheduler:
\n image: apache\/airflow:2.5.0
\n environment:
\n – AIRFLOW__CORE__EXECUTOR=CeleryExecutor
\n – AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2:\/\/airflow:airflow@airflow-db:5432\/airflow
\n – AIRFLOW__CELERY__RESULT_BACKEND=db+postgresql:\/\/airflow:airflow@airflow-db:5432\/airflow
\n – AIRFLOW__CELERY__BROKER_URL=redis:\/\/:@redis:6379\/0
\n volumes:
\n – .\/airflow\/dags:\/opt\/airflow\/dags
\n – .\/airflow\/logs:\/opt\/airflow\/logs
\n depends_on:
\n – airflow-db
\n – redis
\n networks:
\n – ml-network<\/p>\n

# \u65e5\u5fd7\u6536\u96c6 (ELK Stack)
\n elasticsearch:
\n image: elasticsearch:8.6.0
\n environment:
\n – discovery.type=single-node
\n – xpack.security.enabled=false
\n volumes:
\n – elasticsearch_data:\/usr\/share\/elasticsearch\/data
\n ports:
\n – "9200:9200"
\n networks:
\n – ml-network<\/p>\n

logstash:
\n image: logstash:8.6.0
\n volumes:
\n – .\/logstash\/logstash.conf:\/usr\/share\/logstash\/pipeline\/logstash.conf
\n depends_on:
\n – elasticsearch
\n networks:
\n – ml-network<\/p>\n

kibana:
\n image: kibana:8.6.0
\n environment:
\n – ELASTICSEARCH_HOSTS=http:\/\/elasticsearch:9200
\n ports:
\n – "5601:5601"
\n depends_on:
\n – elasticsearch
\n networks:
\n – ml-network<\/p>\n

networks:
\n ml-network:
\n driver: bridge<\/p>\n

volumes:
\n postgres_data:
\n redis_data:
\n prometheus_data:
\n grafana_data:
\n elasticsearch_data: <\/p>\n

3.2 Kubernetes\u90e8\u7f72\u914d\u7f6e<\/h4>\n

yaml<\/p>\n<\/p>\n

\u590d\u5236<\/p>\n<\/p>\n

\u4e0b\u8f7d<\/p>\n

# k8s-deployment.yaml
\napiVersion: apps\/v1
\nkind: Deployment
\nmetadata:
\n name: ml-model-api
\n namespace: ml-production
\n labels:
\n app: ml-model-api
\n component: model-serving
\nspec:
\n replicas: 3
\n selector:
\n matchLabels:
\n app: ml-model-api
\n template:
\n metadata:
\n labels:
\n app: ml-model-api
\n component: model-serving
\n annotations:
\n prometheus.io\/scrape: "true"
\n prometheus.io\/port: "8000"
\n prometheus.io\/path: "\/metrics"
\n spec:
\n serviceAccountName: ml-service-account
\n containers:
\n – name: model-api
\n image: registry.example.com\/ml-model-api:v1.2.0
\n imagePullPolicy: IfNotPresent
\n ports:
\n – containerPort: 8000
\n name: http
\n env:
\n – name: ENVIRONMENT
\n value: "production"
\n – name: LOG_LEVEL
\n value: "INFO"
\n – name: MODEL_CONFIG_PATH
\n value: "\/app\/config\/models.json"
\n – name: DATABASE_URL
\n valueFrom:
\n secretKeyRef:
\n name: ml-secrets
\n key: database-url
\n – name: REDIS_HOST
\n value: "redis-master.redis.svc.cluster.local"
\n – name: PROMETHEUS_PUSH_GATEWAY
\n value: "prometheus-pushgateway.monitoring.svc.cluster.local:9091"
\n resources:
\n requests:
\n memory: "1Gi"
\n cpu: "500m"
\n limits:
\n memory: "4Gi"
\n cpu: "2000m"
\n volumeMounts:
\n – name: model-storage
\n mountPath: \/app\/models
\n readOnly: true
\n – name: config-volume
\n mountPath: \/app\/config
\n – name: logs-volume
\n mountPath: \/app\/logs
\n livenessProbe:
\n httpGet:
\n path: \/health
\n port: 8000
\n initialDelaySeconds: 30
\n periodSeconds: 10
\n timeoutSeconds: 5
\n failureThreshold: 3
\n readinessProbe:
\n httpGet:
\n path: \/health
\n port: 8000
\n initialDelaySeconds: 5
\n periodSeconds: 5
\n timeoutSeconds: 3
\n startupProbe:
\n httpGet:
\n path: \/health
\n port: 8000
\n failureThreshold: 30
\n periodSeconds: 10
\n volumes:
\n – name: model-storage
\n persistentVolumeClaim:
\n claimName: model-pvc
\n – name: config-volume
\n configMap:
\n name: ml-config
\n – name: logs-volume
\n emptyDir: {}
\n affinity:
\n podAntiAffinity:
\n preferredDuringSchedulingIgnoredDuringExecution:
\n – weight: 100
\n podAffinityTerm:
\n labelSelector:
\n matchExpressions:
\n – key: app
\n operator: In
\n values:
\n – ml-model-api
\n topologyKey: kubernetes.io\/hostname
\n nodeSelector:
\n node-type: model-serving
\n tolerations:
\n – key: "model-serving"
\n operator: "Equal"
\n value: "true"
\n effect: "NoSchedule"
\n—
\n# Horizontal Pod Autoscaler
\napiVersion: autoscaling\/v2
\nkind: HorizontalPodAutoscaler
\nmetadata:
\n name: ml-model-api-hpa
\n namespace: ml-production
\nspec:
\n scaleTargetRef:
\n apiVersion: apps\/v1
\n kind: Deployment
\n name: ml-model-api
\n minReplicas: 3
\n maxReplicas: 10
\n metrics:
\n – type: Resource
\n resource:
\n name: cpu
\n target:
\n type: Utilization
\n averageUtilization: 70
\n – type: Resource
\n resource:
\n name: memory
\n target:
\n type: Utilization
\n averageUtilization: 80
\n – type: Pods
\n pods:
\n metric:
\n name: model_requests_per_second
\n target:
\n type: AverageValue
\n averageValue: 1000
\n behavior:
\n scaleDown:
\n stabilizationWindowSeconds: 300
\n policies:
\n – type: Percent
\n value: 10
\n periodSeconds: 60
\n scaleUp:
\n stabilizationWindowSeconds: 60
\n policies:
\n – type: Percent
\n value: 100
\n periodSeconds: 60
\n—
\n# Service
\napiVersion: v1
\nkind: Service
\nmetadata:
\n name: ml-model-api
\n namespace: ml-production
\n annotations:
\n prometheus.io\/scrape: "true"
\n prometheus.io\/port: "8000"
\nspec:
\n selector:
\n app: ml-model-api
\n ports:
\n – name: http
\n port: 8000
\n targetPort: 8000
\n type: ClusterIP
\n—
\n# Ingress
\napiVersion: networking.k8s.io\/v1
\nkind: Ingress
\nmetadata:
\n name: ml-model-api-ingress
\n namespace: ml-production
\n annotations:
\n nginx.ingress.kubernetes.io\/rewrite-target: \/
\n nginx.ingress.kubernetes.io\/ssl-redirect: "true"
\n nginx.ingress.kubernetes.io\/proxy-body-size: "50m"
\n nginx.ingress.kubernetes.io\/proxy-read-timeout: "300"
\n nginx.ingress.kubernetes.io\/proxy-send-timeout: "300"
\nspec:
\n ingressClassName: nginx
\n tls:
\n – hosts:
\n – ml-api.example.com
\n secretName: ml-api-tls
\n rules:
\n – host: ml-api.example.com
\n http:
\n paths:
\n – path: \/
\n pathType: Prefix
\n backend:
\n service:
\n name: ml-model-api
\n port:
\n number: 8000
\n – path: \/metrics
\n pathType: Prefix
\n backend:
\n service:
\n name: ml-model-api
\n port:
\n number: 8000
\n—
\n# ConfigMap
\napiVersion: v1
\nkind: ConfigMap
\nmetadata:
\n name: ml-config
\n namespace: ml-production
\ndata:
\n models.json: |
\n {
\n "models": [
\n {
\n "name": "recommendation",
\n "version": "v1",
\n "framework": "tensorflow",
\n "model_path": "\/app\/models\/recommendation\/v1\/model.h5",
\n "metadata_path": "\/app\/models\/recommendation\/v1\/metadata.json",
\n "min_memory_mb": 512,
\n "max_batch_size": 100,
\n "feature_pipeline": "recommendation_features"
\n },
\n {
\n "name": "fraud_detection",
\n "version": "v2",
\n "framework": "pytorch",
\n "model_path": "\/app\/models\/fraud\/v2\/model.pt",
\n "metadata_path": "\/app\/models\/fraud\/v2\/metadata.json",
\n "min_memory_mb": 1024,
\n "max_batch_size": 50
\n }
\n ],
\n "feature_pipelines": {
\n "recommendation_features": {
\n "pipeline": "user_item_features",
\n "version": "v1"
\n }
\n },
\n "performance": {
\n "default_timeout_ms": 5000,
\n "max_concurrent_requests": 100,
\n "circuit_breaker_threshold": 0.5
\n }
\n }<\/p>\n

ab_test.json: |
\n {
\n "experiments": [
\n {
\n "name": "recommendation_model_v2",
\n "description": "\u6d4b\u8bd5\u65b0\u7248\u63a8\u8350\u6a21\u578b\u7684\u6548\u679c",
\n "status": "running",
\n "groups": [
\n {
\n "name": "control",
\n "weight": 0.5,
\n "model_name": "recommendation",
\n "model_version": "v1"
\n },
\n {
\n "name": "treatment_v2",
\n "weight": 0.3,
\n "model_name": "recommendation",
\n "model_version": "v2"
\n },
\n {
\n "name": "treatment_v3",
\n "weight": 0.2,
\n "model_name": "recommendation",
\n "model_version": "v3"
\n }
\n ],
\n "assignment_algorithm": "hash_based",
\n "sample_rate": 0.1,
\n "metrics": ["click_rate", "purchase_rate", "revenue"],
\n "alert_rules": {
\n "sample_size": {
\n "min_users": 1000,
\n "warning_threshold": 100
\n },
\n "safety": {
\n "max_deterioration": -0.1
\n }
\n }
\n }
\n ]
\n }
\n—
\n# PersistentVolumeClaim for models
\napiVersion: v1
\nkind: PersistentVolumeClaim
\nmetadata:
\n name: model-pvc
\n namespace: ml-production
\nspec:
\n accessModes:
\n – ReadOnlyMany
\n resources:
\n requests:
\n storage: 100Gi
\n storageClassName: ssd
\n—
\n# ServiceAccount and RBAC
\napiVersion: v1
\nkind: ServiceAccount
\nmetadata:
\n name: ml-service-account
\n namespace: ml-production
\n—
\napiVersion: rbac.authorization.k8s.io\/v1
\nkind: Role
\nmetadata:
\n name: ml-service-role
\n namespace: ml-production
\nrules:
\n– apiGroups: [""]
\n resources: ["configmaps"]
\n verbs: ["get", "list", "watch"]
\n– apiGroups: [""]
\n resources: ["pods", "services"]
\n verbs: ["get", "list"]
\n—
\napiVersion: rbac.authorization.k8s.io\/v1
\nkind: RoleBinding
\nmetadata:
\n name: ml-service-role-binding
\n namespace: ml-production
\nroleRef:
\n apiGroup: rbac.authorization.k8s.io
\n kind: Role
\n name: ml-service-role
\nsubjects:
\n– kind: ServiceAccount
\n name: ml-service-account
\n namespace: ml-production <\/p>\n

\u200b<\/p>\n

\u7bc7\u5e45\u9650\u5236\u4e0b\u9762\u5c31\u53ea\u80fd\u7ed9\u5927\u5bb6\u5c55\u793a\u5c0f\u518c\u90e8\u5206\u5185\u5bb9\u4e86\u3002\u6574\u7406\u4e86\u4e00\u4efd\u6838\u5fc3\u9762\u8bd5\u7b14\u8bb0\u5305\u62ec\u4e86：Java\u9762\u8bd5\u3001Spring\u3001JVM\u3001MyBatis\u3001Redis\u3001MySQL\u3001\u5e76\u53d1\u7f16\u7a0b\u3001\u5fae\u670d\u52a1\u3001Linux\u3001Springboot\u3001SpringCloud\u3001MQ\u3001Kafc<\/p>\n

\u9700\u8981\u5168\u5957\u9762\u8bd5\u7b14\u8bb0\u53ca\u7b54\u6848 \u3010\u70b9\u51fb\u6b64\u5904\u5373\u53ef\/\u514d\u8d39\u83b7\u53d6\u3011\u200b\u200b\u200b<\/span><\/p>\n

\u200b\u200b\u200b\u200b\u200b\u200b\u200b\u200b\u200b\u200b\u200b\u200b<\/p>\n<\/p>\n

3.3 \u6027\u80fd\u4f18\u5316\u4e0e\u76d1\u63a7<\/h4>\n

python<\/p>\n<\/p>\n

\u590d\u5236<\/p>\n<\/p>\n

\u4e0b\u8f7d<\/p>\n

"""
\n\u6027\u80fd\u4f18\u5316\u4e0e\u76d1\u63a7\u7cfb\u7edf
\n"""
\nimport asyncio
\nimport time
\nfrom dataclasses import dataclass, field
\nfrom datetime import datetime, timedelta
\nfrom typing import Dict, List, Optional, Any, Callable
\nimport threading
\nfrom concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
\nimport psutil
\nimport gc
\nfrom prometheus_client import start_http_server, Summary, Counter, Gauge, Histogram
\nimport prometheus_client
\nfrom circuitbreaker import circuit
\nimport backoff
\nimport redis
\nfrom functools import wraps, lru_cache
\nimport numpy as np
\nfrom contextlib import contextmanager<\/p>\n

# ==================== \u6027\u80fd\u76d1\u63a7\u88c5\u9970\u5668 ====================
\ndef monitor_performance(name: str):
\n """\u6027\u80fd\u76d1\u63a7\u88c5\u9970\u5668"""
\n def decorator(func):
\n @wraps(func)
\n async def async_wrapper(*args, **kwargs):
\n start_time = time.time()
\n try:
\n # \u8bb0\u5f55\u8bf7\u6c42\u5f00\u59cb
\n REQUEST_COUNTER.labels(function=name).inc()<\/p>\n

result = await func(*args, **kwargs)<\/p>\n

# \u8bb0\u5f55\u6210\u529f
\n REQUEST_DURATION.labels(function=name).observe(time.time() – start_time)
\n REQUEST_SUCCESS_COUNTER.labels(function=name).inc()<\/p>\n

return result
\n except Exception as e:
\n # \u8bb0\u5f55\u5931\u8d25
\n REQUEST_ERROR_COUNTER.labels(function=name, error_type=type(e).__name__).inc()
\n raise
\n finally:
\n # \u8bb0\u5f55\u5185\u5b58\u4f7f\u7528
\n MEMORY_USAGE.labels(function=name).set(psutil.Process().memory_info().rss)<\/p>\n

@wraps(func)
\n def sync_wrapper(*args, **kwargs):
\n start_time = time.time()
\n try:
\n REQUEST_COUNTER.labels(function=name).inc()<\/p>\n

result = func(*args, **kwargs)<\/p>\n

REQUEST_DURATION.labels(function=name).observe(time.time() – start_time)
\n REQUEST_SUCCESS_COUNTER.labels(function=name).inc()<\/p>\n

return result
\n except Exception as e:
\n REQUEST_ERROR_COUNTER.labels(function=name, error_type=type(e).__name__).inc()
\n raise
\n finally:
\n MEMORY_USAGE.labels(function=name).set(psutil.Process().memory_info().rss)<\/p>\n

if asyncio.iscoroutinefunction(func):
\n return async_wrapper
\n else:
\n return sync_wrapper<\/p>\n

return decorator<\/p>\n

# ==================== \u65ad\u8def\u5668\u6a21\u5f0f ====================
\nclass CircuitBreaker:
\n """\u65ad\u8def\u5668\u5b9e\u73b0"""<\/p>\n

def __init__(
\n self,
\n failure_threshold: int = 5,
\n recovery_timeout: int = 60,
\n expected_exceptions: tuple = (Exception,)
\n ):
\n self.failure_threshold = failure_threshold
\n self.recovery_timeout = recovery_timeout
\n self.expected_exceptions = expected_exceptions<\/p>\n

self.failures = 0
\n self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
\n self.last_failure_time = None
\n self.metrics = {
\n 'total_calls': 0,
\n 'successful_calls': 0,
\n 'failed_calls': 0,
\n 'circuit_opened': 0
\n }<\/p>\n

@contextmanager
\n def protect(self):
\n """\u4fdd\u62a4\u4ee3\u7801\u5757"""
\n self.metrics['total_calls'] += 1<\/p>\n

# \u68c0\u67e5\u65ad\u8def\u5668\u72b6\u6001
\n if self.state == "OPEN":
\n if self._can_attempt_recovery():
\n self.state = "HALF_OPEN"
\n else:
\n self.metrics['circuit_opened'] += 1
\n raise CircuitBreakerOpenError("Circuit breaker is OPEN")<\/p>\n

try:
\n yield
\n self._on_success()<\/p>\n

except self.expected_exceptions as e:
\n self._on_failure()
\n raise<\/p>\n

def _can_attempt_recovery(self) -> bool:
\n """\u68c0\u67e5\u662f\u5426\u53ef\u4ee5\u5c1d\u8bd5\u6062\u590d"""
\n if self.last_failure_time is None:
\n return True<\/p>\n

elapsed = time.time() – self.last_failure_time
\n return elapsed >= self.recovery_timeout<\/p>\n

def _on_success(self):
\n """\u6210\u529f\u65f6\u5904\u7406"""
\n self.metrics['successful_calls'] += 1<\/p>\n

if self.state == "HALF_OPEN":
\n self.state = "CLOSED"
\n self.failures = 0<\/p>\n

def _on_failure(self):
\n """\u5931\u8d25\u65f6\u5904\u7406"""
\n self.metrics['failed_calls'] += 1
\n self.failures += 1
\n self.last_failure_time = time.time()<\/p>\n

if self.failures >= self.failure_threshold:
\n self.state = "OPEN"<\/p>\n

class CircuitBreakerOpenError(Exception):
\n """\u65ad\u8def\u5668\u6253\u5f00\u5f02\u5e38"""
\n pass<\/p>\n

# ==================== \u7f13\u5b58\u4f18\u5316 ====================
\nclass ModelPredictionCache:
\n """\u6a21\u578b\u9884\u6d4b\u7f13\u5b58"""<\/p>\n

def __init__(self, redis_client=None, ttl: int = 300):
\n self.redis_client = redis_client
\n self.ttl = ttl # \u7f13\u5b58\u65f6\u95f4（\u79d2）
\n self.local_cache = {}
\n self.local_cache_ttl = {}
\n self.hits = 0
\n self.misses = 0<\/p>\n

def get_cache_key(self, model_name: str, model_version: str, features_hash: str) -> str:
\n """\u751f\u6210\u7f13\u5b58\u952e"""
\n return f"prediction:{model_name}:{model_version}:{features_hash}"<\/p>\n

def get(self, model_name: str, model_version: str, features: Dict) -> Optional[Any]:
\n """\u83b7\u53d6\u7f13\u5b58"""
\n features_hash = self._hash_features(features)
\n cache_key = self.get_cache_key(model_name, model_version, features_hash)<\/p>\n

# 1. \u68c0\u67e5\u672c\u5730\u7f13\u5b58
\n if cache_key in self.local_cache:
\n ttl = self.local_cache_ttl.get(cache_key, 0)
\n if ttl > time.time():
\n self.hits += 1
\n return self.local_cache[cache_key]<\/p>\n

# 2. \u68c0\u67e5Redis\u7f13\u5b58
\n if self.redis_client:
\n try:
\n cached = self.redis_client.get(cache_key)
\n if cached:
\n # \u89e3\u6790\u7f13\u5b58\u6570\u636e
\n result = json.loads(cached)<\/p>\n

# \u66f4\u65b0\u672c\u5730\u7f13\u5b58
\n self.local_cache[cache_key] = result
\n self.local_cache_ttl[cache_key] = time.time() + min(self.ttl, 30)<\/p>\n

self.hits += 1
\n return result
\n except Exception as e:
\n # Redis\u8bbf\u95ee\u5931\u8d25，\u7ee7\u7eed\u6267\u884c
\n pass<\/p>\n

self.misses += 1
\n return None<\/p>\n

def set(self, model_name: str, model_version: str, features: Dict, result: Any):
\n """\u8bbe\u7f6e\u7f13\u5b58"""
\n features_hash = self._hash_features(features)
\n cache_key = self.get_cache_key(model_name, model_version, features_hash)<\/p>\n

# 1. \u8bbe\u7f6e\u672c\u5730\u7f13\u5b58
\n self.local_cache[cache_key] = result
\n self.local_cache_ttl[cache_key] = time.time() + min(self.ttl, 30)<\/p>\n

# 2. \u8bbe\u7f6eRedis\u7f13\u5b58
\n if self.redis_client:
\n try:
\n self.redis_client.setex(
\n cache_key,
\n self.ttl,
\n json.dumps(result, default=str)
\n )
\n except Exception as e:
\n # Redis\u8bbf\u95ee\u5931\u8d25，\u5ffd\u7565
\n pass<\/p>\n

def _hash_features(self, features: Dict) -> str:
\n """\u54c8\u5e0c\u7279\u5f81"""
\n import hashlib
\n features_str = json.dumps(features, sort_keys=True)
\n return hashlib.md5(features_str.encode()).hexdigest()<\/p>\n

def get_stats(self) -> Dict[str, Any]:
\n """\u83b7\u53d6\u7f13\u5b58\u7edf\u8ba1"""
\n total = self.hits + self.misses
\n hit_rate = self.hits \/ total if total > 0 else 0<\/p>\n

return {
\n 'hits': self.hits,
\n 'misses': self.misses,
\n 'total': total,
\n 'hit_rate': hit_rate,
\n 'local_cache_size': len(self.local_cache)
\n }<\/p>\n

def clear(self):
\n """\u6e05\u7a7a\u7f13\u5b58"""
\n self.local_cache.clear()
\n self.local_cache_ttl.clear()
\n self.hits = 0
\n self.misses = 0<\/p>\n

# ==================== \u6279\u91cf\u5904\u7406\u4f18\u5316 ====================
\nclass BatchProcessor:
\n """\u6279\u91cf\u5904\u7406\u5668"""<\/p>\n

def __init__(self, max_batch_size: int = 100, max_wait_time: float = 0.1):
\n self.max_batch_size = max_batch_size
\n self.max_wait_time = max_wait_time<\/p>\n

self.queue = asyncio.Queue()
\n self.processing = False
\n self.processing_task = None<\/p>\n

# \u7edf\u8ba1\u4fe1\u606f
\n self.total_batches = 0
\n self.total_items = 0
\n self.average_batch_size = 0<\/p>\n

async def add_item(self, item: Any) -> asyncio.Future:
\n """\u6dfb\u52a0\u9879\u76ee\u5230\u6279\u91cf\u5904\u7406\u5668"""
\n future = asyncio.Future()
\n await self.queue.put((item, future))<\/p>\n

# \u5982\u679c\u6ca1\u6709\u5728\u5904\u7406，\u542f\u52a8\u5904\u7406\u4efb\u52a1
\n if not self.processing:
\n self.start_processing()<\/p>\n

return future<\/p>\n

def start_processing(self):
\n """\u542f\u52a8\u5904\u7406"""
\n if self.processing:
\n return<\/p>\n

self.processing = True
\n self.processing_task = asyncio.create_task(self._process_batches())<\/p>\n

async def _process_batches(self):
\n """\u5904\u7406\u6279\u6b21"""
\n while self.processing:
\n try:
\n # \u6536\u96c6\u4e00\u6279\u9879\u76ee
\n batch = []
\n start_time = time.time()<\/p>\n

# \u6536\u96c6\u76f4\u5230\u8fbe\u5230\u6700\u5927\u6279\u6b21\u5927\u5c0f\u6216\u8d85\u65f6
\n while len(batch) < self.max_batch_size:
\n try:
\n item, future = await asyncio.wait_for(
\n self.queue.get(),
\n timeout=self.max_wait_time
\n )
\n batch.append((item, future))
\n except asyncio.TimeoutError:
\n # \u8d85\u65f6，\u5904\u7406\u5f53\u524d\u6279\u6b21
\n break<\/p>\n

if not batch:
\n # \u961f\u5217\u4e3a\u7a7a，\u505c\u6b62\u5904\u7406
\n self.processing = False
\n break<\/p>\n

# \u5904\u7406\u6279\u6b21
\n await self._process_batch(batch)<\/p>\n

# \u66f4\u65b0\u7edf\u8ba1
\n self.total_batches += 1
\n self.total_items += len(batch)
\n self.average_batch_size = self.total_items \/ self.total_batches<\/p>\n

except Exception as e:
\n logger.error(f"Batch processing error: {e}")
\n # \u8bbe\u7f6e\u6240\u6709\u672a\u5b8c\u6210\u7684future\u4e3a\u5f02\u5e38
\n for _, future in batch:
\n if not future.done():
\n future.set_exception(e)<\/p>\n

self.processing = False<\/p>\n

async def _process_batch(self, batch: List[tuple]):
\n """\u5904\u7406\u5355\u4e2a\u6279\u6b21"""
\n # \u63d0\u53d6\u9879\u76ee\u548cfuture
\n items = [item for item, _ in batch]
\n futures = [future for _, future in batch]<\/p>\n

try:
\n # \u6267\u884c\u6279\u91cf\u5904\u7406
\n results = await self.process_items(items)<\/p>\n

# \u8bbe\u7f6efuture\u7ed3\u679c
\n for future, result in zip(futures, results):
\n if not future.done():
\n future.set_result(result)<\/p>\n

except Exception as e:
\n # \u8bbe\u7f6e\u6240\u6709future\u4e3a\u5f02\u5e38
\n for future in futures:
\n if not future.done():
\n future.set_exception(e)<\/p>\n

async def process_items(self, items: List[Any]) -> List[Any]:
\n """\u5904\u7406\u9879\u76ee（\u5b50\u7c7b\u91cd\u5199）"""
\n raise NotImplementedError<\/p>\n

def get_stats(self) -> Dict[str, Any]:
\n """\u83b7\u53d6\u7edf\u8ba1\u4fe1\u606f"""
\n return {
\n 'queue_size': self.queue.qsize(),
\n 'processing': self.processing,
\n 'total_batches': self.total_batches,
\n 'total_items': self.total_items,
\n 'average_batch_size': self.average_batch_size,
\n 'max_batch_size': self.max_batch_size,
\n 'max_wait_time': self.max_wait_time
\n }<\/p>\n

class ModelBatchProcessor(BatchProcessor):
\n """\u6a21\u578b\u6279\u91cf\u5904\u7406\u5668"""<\/p>\n

def __init__(self, model, max_batch_size: int = 100, max_wait_time: float = 0.1):
\n super().__init__(max_batch_size, max_wait_time)
\n self.model = model<\/p>\n

async def process_items(self, items: List[Dict]) -> List[Dict]:
\n """\u6279\u91cf\u5904\u7406\u6a21\u578b\u9884\u6d4b"""
\n try:
\n # \u6279\u91cf\u9884\u6d4b
\n predictions = await self.model.batch_predict(items)
\n return predictions<\/p>\n

except Exception as e:
\n logger.error(f"Batch prediction error: {e}")
\n raise<\/p>\n

# ==================== \u5185\u5b58\u7ba1\u7406\u4f18\u5316 ====================
\nclass MemoryManager:
\n """\u5185\u5b58\u7ba1\u7406\u5668"""<\/p>\n

def __init__(self, max_memory_usage: float = 0.8):
\n self.max_memory_usage = max_memory_usage # \u6700\u5927\u5185\u5b58\u4f7f\u7528\u7387
\n self.memory_warnings = []
\n self.last_gc_time = time.time()
\n self.gc_interval = 300 # \u6bcf5\u5206\u949f\u5f3a\u5236GC\u4e00\u6b21<\/p>\n

# \u76d1\u63a7\u6307\u6807
\n self.memory_usage_gauge = Gauge('memory_usage_percent', 'Memory usage percentage')
\n self.gc_count_gauge = Gauge('gc_collections', 'Garbage collection count')<\/p>\n

def check_memory_usage(self) -> Dict[str, Any]:
\n """\u68c0\u67e5\u5185\u5b58\u4f7f\u7528\u60c5\u51b5"""
\n process = psutil.Process()
\n memory_info = process.memory_info()<\/p>\n

# \u83b7\u53d6\u7cfb\u7edf\u5185\u5b58\u4fe1\u606f
\n system_memory = psutil.virtual_memory()<\/p>\n

# \u8ba1\u7b97\u4f7f\u7528\u7387
\n process_usage = memory_info.rss \/ system_memory.total
\n system_usage = system_memory.used \/ system_memory.total<\/p>\n

# \u66f4\u65b0\u76d1\u63a7\u6307\u6807
\n self.memory_usage_gauge.set(system_usage * 100)<\/p>\n

result = {
\n 'process_memory_mb': memory_info.rss \/ 1024 \/ 1024,
\n 'process_memory_percent': process_usage * 100,
\n 'system_memory_percent': system_usage * 100,
\n 'system_memory_available_mb': system_memory.available \/ 1024 \/ 1024,
\n 'memory_warnings': len(self.memory_warnings)
\n }<\/p>\n

# \u68c0\u67e5\u662f\u5426\u9700\u8981GC
\n if system_usage > self.max_memory_usage:
\n warning = {
\n 'timestamp': datetime.now().isoformat(),
\n 'memory_usage': system_usage,
\n 'threshold': self.max_memory_usage,
\n 'message': 'High memory usage detected'
\n }
\n self.memory_warnings.append(warning)<\/p>\n

# \u5f3a\u5236GC
\n self.force_gc()<\/p>\n

# \u9650\u5236\u8b66\u544a\u6570\u91cf
\n if len(self.memory_warnings) > 100:
\n self.memory_warnings = self.memory_warnings[-100:]<\/p>\n

# \u5b9a\u671fGC
\n if time.time() – self.last_gc_time > self.gc_interval:
\n self.force_gc()<\/p>\n

return result<\/p>\n

def force_gc(self):
\n """\u5f3a\u5236\u5783\u573e\u56de\u6536"""
\n gc.collect()
\n self.last_gc_time = time.time()<\/p>\n

# \u66f4\u65b0\u76d1\u63a7\u6307\u6807
\n self.gc_count_gauge.inc()<\/p>\n

def get_memory_stats(self) -> Dict[str, Any]:
\n """\u83b7\u53d6\u5185\u5b58\u7edf\u8ba1"""
\n process = psutil.Process()<\/p>\n

return {
\n 'timestamp': datetime.now().isoformat(),
\n 'process': {
\n 'rss_mb': process.memory_info().rss \/ 1024 \/ 1024,
\n 'vms_mb': process.memory_info().vms \/ 1024 \/ 1024,
\n 'percent': process.memory_percent(),
\n 'threads': process.num_threads()
\n },
\n 'system': {
\n 'total_mb': psutil.virtual_memory().total \/ 1024 \/ 1024,
\n 'available_mb': psutil.virtual_memory().available \/ 1024 \/ 1024,
\n 'percent': psutil.virtual_memory().percent,
\n 'swap_percent': psutil.swap_memory().percent if hasattr(psutil, 'swap_memory') else 0
\n },
\n 'warnings': self.memory_warnings[-10:] if self.memory_warnings else []
\n }<\/p>\n

# ==================== \u8d1f\u8f7d\u5747\u8861\u4f18\u5316 ====================
\nclass LoadBalancer:
\n """\u8d1f\u8f7d\u5747\u8861\u5668"""<\/p>\n

def __init__(self, endpoints: List[str]):
\n self.endpoints = endpoints
\n self.current_index = 0
\n self.endpoint_stats = {endpoint: {'success': 0, 'failure': 0, 'latency': []} for endpoint in endpoints}
\n self.lock = threading.Lock()<\/p>\n

# \u5065\u5eb7\u68c0\u67e5\u95f4\u9694
\n self.health_check_interval = 30
\n self.unhealthy_endpoints = set()<\/p>\n

# \u542f\u52a8\u5065\u5eb7\u68c0\u67e5
\n self.health_check_thread = threading.Thread(target=self._health_check_loop, daemon=True)
\n self.health_check_thread.start()<\/p>\n

def get_endpoint(self) -> Optional[str]:
\n """\u83b7\u53d6\u4e00\u4e2a\u53ef\u7528\u7684\u7aef\u70b9"""
\n with self.lock:
\n # \u8fc7\u6ee4\u6389\u4e0d\u5065\u5eb7\u7684\u7aef\u70b9
\n healthy_endpoints = [e for e in self.endpoints if e not in self.unhealthy_endpoints]<\/p>\n

if not healthy_endpoints:
\n return None<\/p>\n

# \u4f7f\u7528\u8f6e\u8be2\u7b97\u6cd5
\n endpoint = healthy_endpoints[self.current_index % len(healthy_endpoints)]
\n self.current_index += 1<\/p>\n

return endpoint<\/p>\n

def record_success(self, endpoint: str, latency: float):
\n """\u8bb0\u5f55\u6210\u529f"""
\n with self.lock:
\n if endpoint in self.endpoint_stats:
\n self.endpoint_stats[endpoint]['success'] += 1
\n self.endpoint_stats[endpoint]['latency'].append(latency)<\/p>\n

# \u9650\u5236\u5ef6\u8fdf\u8bb0\u5f55\u6570\u91cf
\n if len(self.endpoint_stats[endpoint]['latency']) > 1000:
\n self.endpoint_stats[endpoint]['latency'] = self.endpoint_stats[endpoint]['latency'][-1000:]<\/p>\n

def record_failure(self, endpoint: str):
\n """\u8bb0\u5f55\u5931\u8d25"""
\n with self.lock:
\n if endpoint in self.endpoint_stats:
\n self.endpoint_stats[endpoint]['failure'] += 1<\/p>\n

def _health_check_loop(self):
\n """\u5065\u5eb7\u68c0\u67e5\u5faa\u73af"""
\n while True:
\n time.sleep(self.health_check_interval)
\n self._check_endpoints_health()<\/p>\n

def _check_endpoints_health(self):
\n """\u68c0\u67e5\u7aef\u70b9\u5065\u5eb7\u72b6\u6001"""
\n for endpoint in self.endpoints:
\n is_healthy = self._check_endpoint_health(endpoint)<\/p>\n

with self.lock:
\n if is_healthy and endpoint in self.unhealthy_endpoints:
\n self.unhealthy_endpoints.remove(endpoint)
\n logger.info(f"Endpoint {endpoint} is now healthy")
\n elif not is_healthy and endpoint not in self.unhealthy_endpoints:
\n self.unhealthy_endpoints.add(endpoint)
\n logger.warning(f"Endpoint {endpoint} is now unhealthy")<\/p>\n

def _check_endpoint_health(self, endpoint: str) -> bool:
\n """\u68c0\u67e5\u5355\u4e2a\u7aef\u70b9\u5065\u5eb7\u72b6\u6001"""
\n try:
\n # \u7b80\u5355\u7684HTTP\u5065\u5eb7\u68c0\u67e5
\n import requests
\n response = requests.get(f"{endpoint}\/health", timeout=5)
\n return response.status_code == 200
\n except Exception as e:
\n return False<\/p>\n

def get_stats(self) -> Dict[str, Any]:
\n """\u83b7\u53d6\u7edf\u8ba1\u4fe1\u606f"""
\n with self.lock:
\n stats = {}<\/p>\n

for endpoint, endpoint_stat in self.endpoint_stats.items():
\n total = endpoint_stat['success'] + endpoint_stat['failure']
\n success_rate = endpoint_stat['success'] \/ total if total > 0 else 0<\/p>\n

latency_values = endpoint_stat['latency']
\n avg_latency = sum(latency_values) \/ len(latency_values) if latency_values else 0<\/p>\n

stats[endpoint] = {
\n 'success': endpoint_stat['success'],
\n 'failure': endpoint_stat['failure'],
\n 'total': total,
\n 'success_rate': success_rate,
\n 'avg_latency_ms': avg_latency * 1000,
\n 'is_healthy': endpoint not in self.unhealthy_endpoints
\n }<\/p>\n

return {
\n 'total_endpoints': len(self.endpoints),
\n 'healthy_endpoints': len(self.endpoints) – len(self.unhealthy_endpoints),
\n 'endpoint_stats': stats
\n }<\/p>\n

# ==================== \u76d1\u63a7\u4eea\u8868\u677f ====================
\nclass PerformanceDashboard:
\n """\u6027\u80fd\u76d1\u63a7\u4eea\u8868\u677f"""<\/p>\n

def __init__(self, port: int = 9091):
\n self.port = port
\n self.metrics = {}
\n self.alerts = []
\n self.alert_rules = []<\/p>\n

# \u542f\u52a8Prometheus HTTP\u670d\u52a1\u5668
\n start_http_server(self.port)<\/p>\n

def add_metric(self, name: str, metric_type: str, **kwargs):
\n """\u6dfb\u52a0\u6307\u6807"""
\n if metric_type == 'counter':
\n self.metrics[name] = Counter(name, kwargs.get('description', ''), kwargs.get('labelnames', []))
\n elif metric_type == 'gauge':
\n self.metrics[name] = Gauge(name, kwargs.get('description', ''), kwargs.get('labelnames', []))
\n elif metric_type == 'histogram':
\n self.metrics[name] = Histogram(name, kwargs.get('description', ''), kwargs.get('labelnames', []))
\n elif metric_type == 'summary':
\n self.metrics[name] = Summary(name, kwargs.get('description', ''), kwargs.get('labelnames', []))<\/p>\n

def add_alert_rule(self, name: str, condition: Callable, action: Callable, severity: str = 'warning'):
\n """\u6dfb\u52a0\u62a5\u8b66\u89c4\u5219"""
\n self.alert_rules.append({
\n 'name': name,
\n 'condition': condition,
\n 'action': action,
\n 'severity': severity,
\n 'last_triggered': None
\n })<\/p>\n

async def check_alerts(self):
\n """\u68c0\u67e5\u62a5\u8b66"""
\n for rule in self.alert_rules:
\n try:
\n if rule['condition']():
\n # \u89e6\u53d1\u62a5\u8b66
\n rule['action']()<\/p>\n

alert = {
\n 'name': rule['name'],
\n 'severity': rule['severity'],
\n 'timestamp': datetime.now().isoformat(),
\n 'message': f"Alert {rule['name']} triggered"
\n }<\/p>\n

self.alerts.append(alert)
\n rule['last_triggered'] = datetime.now()<\/p>\n

# \u9650\u5236\u62a5\u8b66\u6570\u91cf
\n if len(self.alerts) > 1000:
\n self.alerts = self.alerts[-1000:]<\/p>\n

except Exception as e:
\n logger.error(f"Error checking alert rule {rule['name']}: {e}")<\/p>\n

def get_dashboard_data(self) -> Dict[str, Any]:
\n """\u83b7\u53d6\u4eea\u8868\u677f\u6570\u636e"""
\n # \u6536\u96c6\u7cfb\u7edf\u6307\u6807
\n cpu_percent = psutil.cpu_percent(interval=1)
\n memory = psutil.virtual_memory()
\n disk = psutil.disk_usage('\/')
\n network = psutil.net_io_counters()<\/p>\n

# \u6536\u96c6\u8fdb\u7a0b\u6307\u6807
\n process = psutil.Process()
\n process_memory = process.memory_info()<\/p>\n

dashboard_data = {
\n 'timestamp': datetime.now().isoformat(),
\n 'system': {
\n 'cpu_percent': cpu_percent,
\n 'memory_percent': memory.percent,
\n 'memory_available_mb': memory.available \/ 1024 \/ 1024,
\n 'disk_percent': disk.percent,
\n 'disk_free_gb': disk.free \/ 1024 \/ 1024 \/ 1024,
\n 'network_bytes_sent': network.bytes_sent,
\n 'network_bytes_recv': network.bytes_recv
\n },
\n 'process': {
\n 'memory_rss_mb': process_memory.rss \/ 1024 \/ 1024,
\n 'memory_percent': process.memory_percent(),
\n 'cpu_percent': process.cpu_percent(),
\n 'threads': process.num_threads(),
\n 'connections': len(process.connections())
\n },
\n 'alerts': self.alerts[-20:] if self.alerts else [],
\n 'active_alerts': len([a for a in self.alerts if
\n datetime.fromisoformat(a['timestamp']) >
\n datetime.now() – timedelta(hours=1)]),
\n 'metrics': self._get_metrics_snapshot()
\n }<\/p>\n

return dashboard_data<\/p>\n

def _get_metrics_snapshot(self) -> Dict[str, Any]:
\n """\u83b7\u53d6\u6307\u6807\u5feb\u7167"""
\n snapshot = {}<\/p>\n

for name, metric in self.metrics.items():
\n if hasattr(metric, '_metrics'):
\n # \u83b7\u53d6\u6307\u6807\u503c
\n metric_data = metric._metrics
\n snapshot[name] = {}<\/p>\n

for labels, value in metric_data.items():
\n snapshot[name][str(labels)] = value._value if hasattr(value, '_value') else str(value)<\/p>\n

return snapshot<\/p>\n

async def start_monitoring(self):
\n """\u542f\u52a8\u76d1\u63a7"""
\n # \u542f\u52a8\u62a5\u8b66\u68c0\u67e5\u5faa\u73af
\n asyncio.create_task(self._monitoring_loop())<\/p>\n

logger.info(f"Performance dashboard started on port {self.port}")<\/p>\n

async def _monitoring_loop(self):
\n """\u76d1\u63a7\u5faa\u73af"""
\n while True:
\n try:
\n await self.check_alerts()
\n await asyncio.sleep(60) # \u6bcf\u5206\u949f\u68c0\u67e5\u4e00\u6b21
\n except Exception as e:
\n logger.error(f"Error in monitoring loop: {e}")
\n await asyncio.sleep(60)<\/p>\n

# ==================== \u4e3b\u7a0b\u5e8f ====================
\nasync def main_optimized():
\n """\u4f18\u5316\u7684\u4e3b\u7a0b\u5e8f"""
\n # \u521d\u59cb\u5316\u65e5\u5fd7
\n logging.basicConfig(
\n level=logging.INFO,
\n format='%(asctime)s – %(name)s – %(levelname)s – %(message)s'
\n )<\/p>\n

logger = logging.getLogger(__name__)<\/p>\n

try:
\n logger.info("Starting optimized ML Model Serving system…")<\/p>\n

# 1. \u521d\u59cb\u5316\u6027\u80fd\u76d1\u63a7\u4eea\u8868\u677f
\n dashboard = PerformanceDashboard(port=9091)<\/p>\n

# \u6dfb\u52a0\u6307\u6807
\n dashboard.add_metric('model_predictions_total', 'counter',
\n description='Total model predictions',
\n labelnames=['model_name', 'model_version'])<\/p>\n

dashboard.add_metric('prediction_latency_seconds', 'histogram',
\n description='Prediction latency in seconds',
\n labelnames=['model_name', 'model_version'])<\/p>\n

dashboard.add_metric('active_connections', 'gauge',
\n description='Active connections')<\/p>\n

# \u6dfb\u52a0\u62a5\u8b66\u89c4\u5219
\n dashboard.add_alert_rule(
\n name='high_cpu_usage',
\n condition=lambda: psutil.cpu_percent() > 80,
\n action=lambda: logger.warning("High CPU usage detected"),
\n severity='warning'
\n )<\/p>\n

dashboard.add_alert_rule(
\n name='high_memory_usage',
\n condition=lambda: psutil.virtual_memory().percent > 85,
\n action=lambda: logger.error("High memory usage detected"),
\n severity='critical'
\n )<\/p>\n

# 2. \u521d\u59cb\u5316\u5185\u5b58\u7ba1\u7406\u5668
\n memory_manager = MemoryManager(max_memory_usage=0.8)<\/p>\n

# 3. \u521d\u59cb\u5316\u7f13\u5b58
\n redis_client = redis.Redis(host='localhost', port=6379, db=0)
\n prediction_cache = ModelPredictionCache(redis_client, ttl=300)<\/p>\n

# 4. \u521d\u59cb\u5316\u8d1f\u8f7d\u5747\u8861\u5668
\n endpoints = [
\n "http:\/\/model-service-1:8000",
\n "http:\/\/model-service-2:8000",
\n "http:\/\/model-service-3:8000"
\n ]
\n load_balancer = LoadBalancer(endpoints)<\/p>\n

# 5. \u521d\u59cb\u5316\u65ad\u8def\u5668
\n circuit_breaker = CircuitBreaker(
\n failure_threshold=5,
\n recovery_timeout=60
\n )<\/p>\n

# 6. \u542f\u52a8\u76d1\u63a7
\n await dashboard.start_monitoring()<\/p>\n

logger.info("Optimized ML Model Serving system started successfully")<\/p>\n

# \u4fdd\u6301\u7a0b\u5e8f\u8fd0\u884c
\n while True:
\n # \u5b9a\u671f\u68c0\u67e5\u5185\u5b58\u4f7f\u7528
\n memory_info = memory_manager.check_memory_usage()<\/p>\n

# \u83b7\u53d6\u7f13\u5b58\u7edf\u8ba1
\n cache_stats = prediction_cache.get_stats()<\/p>\n

# \u83b7\u53d6\u8d1f\u8f7d\u5747\u8861\u5668\u7edf\u8ba1
\n lb_stats = load_balancer.get_stats()<\/p>\n

# \u83b7\u53d6\u4eea\u8868\u677f\u6570\u636e
\n dashboard_data = dashboard.get_dashboard_data()<\/p>\n

# \u8bb0\u5f55\u72b6\u6001
\n if time.time() % 60 < 1: # \u6bcf\u5206\u949f\u8bb0\u5f55\u4e00\u6b21
\n logger.info(f"Memory usage: {memory_info['system_memory_percent']:.1f}%")
\n logger.info(f"Cache hit rate: {cache_stats['hit_rate']:.2%}")
\n logger.info(f"Healthy endpoints: {lb_stats['healthy_endpoints']}\/{lb_stats['total_endpoints']}")<\/p>\n

await asyncio.sleep(1)<\/p>\n

except Exception as e:
\n logger.error(f"Error in optimized system: {e}")
\n raise<\/p>\n

# ==================== \u6027\u80fd\u6d4b\u8bd5 ====================
\nclass PerformanceTest:
\n """\u6027\u80fd\u6d4b\u8bd5"""<\/p>\n

def __init__(self, api_url: str):
\n self.api_url = api_url
\n self.results = []<\/p>\n

async def run_concurrent_test(
\n self,
\n num_requests: int,
\n num_concurrent: int,
\n payload: Dict[str, Any]
\n ) -> Dict[str, Any]:
\n """\u8fd0\u884c\u5e76\u53d1\u6d4b\u8bd5"""
\n start_time = time.time()<\/p>\n

# \u521b\u5efa\u8bf7\u6c42\u4efb\u52a1
\n tasks = []
\n for i in range(num_requests):
\n task = asyncio.create_task(self._make_request(payload))
\n tasks.append(task)<\/p>\n

# \u63a7\u5236\u5e76\u53d1\u6570
\n if len(tasks) >= num_concurrent:
\n await asyncio.gather(*tasks)
\n tasks = []<\/p>\n

# \u7b49\u5f85\u5269\u4f59\u4efb\u52a1
\n if tasks:
\n await asyncio.gather(*tasks)<\/p>\n

end_time = time.time()<\/p>\n

# \u5206\u6790\u7ed3\u679c
\n successful = sum(1 for r in self.results if r['success'])
\n failed = len(self.results) – successful<\/p>\n

latencies = [r['latency'] for r in self.results if r['success']]<\/p>\n

if latencies:
\n avg_latency = sum(latencies) \/ len(latencies)
\n p95_latency = np.percentile(latencies, 95)
\n p99_latency = np.percentile(latencies, 99)
\n else:
\n avg_latency = p95_latency = p99_latency = 0<\/p>\n

return {
\n 'total_requests': num_requests,
\n 'concurrent_requests': num_concurrent,
\n 'total_time_seconds': end_time – start_time,
\n 'requests_per_second': num_requests \/ (end_time – start_time) if (end_time – start_time) > 0 else 0,
\n 'successful_requests': successful,
\n 'failed_requests': failed,
\n 'success_rate': successful \/ num_requests if num_requests > 0 else 0,
\n 'average_latency_ms': avg_latency * 1000,
\n 'p95_latency_ms': p95_latency * 1000,
\n 'p99_latency_ms': p99_latency * 1000
\n }<\/p>\n

async def _make_request(self, payload: Dict[str, Any]) -> Dict[str, Any]:
\n """\u53d1\u9001\u5355\u4e2a\u8bf7\u6c42"""
\n request_start = time.time()<\/p>\n

try:
\n async with aiohttp.ClientSession() as session:
\n async with session.post(
\n f"{self.api_url}\/predict",
\n json=payload,
\n timeout=30
\n ) as response:<\/p>\n

if response.status == 200:
\n result = {
\n 'success': True,
\n 'latency': time.time() – request_start,
\n 'status_code': response.status
\n }
\n else:
\n result = {
\n 'success': False,
\n 'latency': time.time() – request_start,
\n 'status_code': response.status,
\n 'error': await response.text()
\n }<\/p>\n

except Exception as e:
\n result = {
\n 'success': False,
\n 'latency': time.time() – request_start,
\n 'error': str(e)
\n }<\/p>\n

self.results.append(result)
\n return result<\/p>\n

def run_load_test_scenarios(self):
\n """\u8fd0\u884c\u8d1f\u8f7d\u6d4b\u8bd5\u573a\u666f"""
\n scenarios = [
\n {'name': 'low_load', 'requests': 100, 'concurrent': 10},
\n {'name': 'medium_load', 'requests': 1000, 'concurrent': 50},
\n {'name': 'high_load', 'requests': 10000, 'concurrent': 100},
\n {'name': 'stress_test', 'requests': 50000, 'concurrent': 200}
\n ]<\/p>\n

all_results = {}<\/p>\n

for scenario in scenarios:
\n logger.info(f"Running {scenario['name']} test…")<\/p>\n

# \u8fd0\u884c\u6d4b\u8bd5
\n asyncio.run(self.run_concurrent_test(
\n scenario['requests'],
\n scenario['concurrent'],
\n self._create_test_payload()
\n ))<\/p>\n

# \u83b7\u53d6\u7ed3\u679c
\n result = self.results[-1] if self.results else {}
\n all_results[scenario['name']] = result<\/p>\n

# \u91cd\u7f6e\u7ed3\u679c
\n self.results = []<\/p>\n

return all_results<\/p>\n

def _create_test_payload(self) -> Dict[str, Any]:
\n """\u521b\u5efa\u6d4b\u8bd5\u8d1f\u8f7d"""
\n return {
\n "model_name": "recommendation",
\n "model_version": "v1",
\n "data": {
\n "user_id": "test_user",
\n "features": [0.1, 0.2, 0.3, 0.4, 0.5] * 20 # 100\u4e2a\u7279\u5f81
\n }
\n }<\/p>\n

def generate_performance_report(self, test_results: Dict[str, Any]) -> str:
\n """\u751f\u6210\u6027\u80fd\u62a5\u544a"""
\n report = "# Performance Test Report\\\\n\\\\n"<\/p>\n

for scenario_name, results in test_results.items():
\n report += f"## {scenario_name.upper()}\\\\n\\\\n"<\/p>\n

report += f"- Total Requests: {results.get('total_requests', 0)}\\\\n"
\n report += f"- Concurrent Requests: {results.get('concurrent_requests', 0)}\\\\n"
\n report += f"- Total Time: {results.get('total_time_seconds', 0):.2f} seconds\\\\n"
\n report += f"- Requests per Second: {results.get('requests_per_second', 0):.2f}\\\\n"
\n report += f"- Success Rate: {results.get('success_rate', 0):.2%}\\\\n"
\n report += f"- Average Latency: {results.get('average_latency_ms', 0):.2f} ms\\\\n"
\n report += f"- P95 Latency: {results.get('p95_latency_ms', 0):.2f} ms\\\\n"
\n report += f"- P99 Latency: {results.get('p99_latency_ms', 0):.2f} ms\\\\n\\\\n"<\/p>\n

# \u6dfb\u52a0\u603b\u7ed3
\n report += "## Summary\\\\n\\\\n"<\/p>\n

# \u5206\u6790\u74f6\u9888
\n bottlenecks = []<\/p>\n

for scenario_name, results in test_results.items():
\n if results.get('success_rate', 1) < 0.95:
\n bottlenecks.append(f"{scenario_name}: Low success rate ({results.get('success_rate', 0):.2%})")<\/p>\n

if results.get('p99_latency_ms', 0) > 1000: # \u8d85\u8fc71\u79d2
\n bottlenecks.append(f"{scenario_name}: High P99 latency ({results.get('p99_latency_ms', 0):.2f} ms)")<\/p>\n

if bottlenecks:
\n report += "### Potential Bottlenecks\\\\n\\\\n"
\n for bottleneck in bottlenecks:
\n report += f"- {bottleneck}\\\\n"
\n report += "\\\\n"
\n else:
\n report += "No significant bottlenecks detected.\\\\n\\\\n"<\/p>\n

# \u5efa\u8bae
\n report += "### Recommendations\\\\n\\\\n"<\/p>\n

max_rps = max(r.get('requests_per_second', 0) for r in test_results.values())
\n if max_rps < 100:
\n report += "- Consider optimizing model inference performance\\\\n"
\n report += "- Implement batching for predictions\\\\n"
\n report += "- Add caching for frequent requests\\\\n"<\/p>\n

return report<\/p>\n

if __name__ == "__main__":
\n # \u8fd0\u884c\u4f18\u5316\u7684\u7cfb\u7edf
\n asyncio.run(main_optimized()) <\/p>\n

\u603b\u7ed3<\/h3>\n

\u672c\u7cfb\u7edf\u5b9e\u73b0\u4e86\u5b8c\u6574\u7684\u673a\u5668\u5b66\u4e60\u6a21\u578b\u5728\u7ebf\u670d\u52a1\u4e0eA\/B\u6d4b\u8bd5\u67b6\u6784，\u5305\u62ec：<\/p>\n

\u6838\u5fc3\u7279\u6027：<\/h4>\n
  • \n

    \u7edf\u4e00\u6a21\u578b\u670d\u52a1\u63a5\u53e3：\u652f\u6301\u591a\u79cd\u6846\u67b6（TensorFlow\u3001PyTorch\u3001Scikit-learn\u7b49）<\/p>\n<\/li>\n

  • \n

    \u9ad8\u6027\u80fdAPI\u670d\u52a1：\u57fa\u4e8eFastAPI\u7684\u5f02\u6b65API，\u652f\u6301\u6279\u91cf\u9884\u6d4b<\/p>\n<\/li>\n

  • \n

    \u5b8c\u6574\u7684A\/B\u6d4b\u8bd5\u7cfb\u7edf：\u5305\u542b\u5b9e\u9a8c\u8bbe\u8ba1\u3001\u7528\u6237\u5206\u914d\u3001\u6548\u679c\u5206\u6790\u7b49\u529f\u80fd<\/p>\n<\/li>\n

  • \n

    \u5b9e\u65f6\u76d1\u63a7\u4e0e\u5206\u6790：\u63d0\u4f9b\u8be6\u7ec6\u7684\u5b9e\u9a8c\u5206\u6790\u548c\u7edf\u8ba1\u68c0\u9a8c<\/p>\n<\/li>\n

  • \n

    \u751f\u4ea7\u7ea7\u90e8\u7f72：\u652f\u6301\u5bb9\u5668\u5316\u548cKubernetes\u90e8\u7f72<\/p>\n<\/li>\n

  • \n

    \u6027\u80fd\u4f18\u5316：\u7f13\u5b58\u3001\u6279\u91cf\u5904\u7406\u3001\u8d1f\u8f7d\u5747\u8861\u3001\u65ad\u8def\u5668\u7b49<\/p>\n<\/li>\n

    \u5173\u952e\u4f18\u52bf：<\/h4>\n
  • \n

    \u53ef\u6269\u5c55\u6027：\u5fae\u670d\u52a1\u67b6\u6784，\u652f\u6301\u6c34\u5e73\u6269\u5c55<\/p>\n<\/li>\n

  • \n

    \u53ef\u9760\u6027：\u5185\u7f6e\u6545\u969c\u6062\u590d\u548c\u5065\u5eb7\u68c0\u67e5\u673a\u5236<\/p>\n<\/li>\n

  • \n

    \u53ef\u89c2\u6d4b\u6027：\u5168\u9762\u7684\u76d1\u63a7\u548c\u65e5\u5fd7\u7cfb\u7edf<\/p>\n<\/li>\n

  • \n

    \u6613\u7528\u6027：RESTful API\u63a5\u53e3，\u6613\u4e8e\u96c6\u6210<\/p>\n<\/li>\n

  • \n

    \u5b89\u5168\u6027：\u652f\u6301\u8ba4\u8bc1\u3001\u6388\u6743\u548cHTTPS<\/p>\n<\/li>\n

    \u751f\u4ea7\u5efa\u8bae：<\/h4>\n
  • \n

    \u6027\u80fd\u8c03\u4f18：\u6839\u636e\u5b9e\u9645\u8d1f\u8f7d\u8c03\u6574\u7f13\u5b58\u7b56\u7565\u548c\u6279\u5904\u7406\u53c2\u6570<\/p>\n<\/li>\n

  • \n

    \u76d1\u63a7\u544a\u8b66：\u8bbe\u7f6e\u5408\u9002\u7684\u9608\u503c，\u53ca\u65f6\u53d1\u73b0\u5e76\u89e3\u51b3\u95ee\u9898<\/p>\n<\/li>\n

  • \n

    \u7248\u672c\u7ba1\u7406：\u5efa\u7acb\u6a21\u578b\u7248\u672c\u63a7\u5236\u6d41\u7a0b<\/p>\n<\/li>\n

  • \n

    \u5b89\u5168\u52a0\u56fa：\u5b9e\u65bdAPI\u7f51\u5173\u3001\u9650\u6d41\u548c\u5ba1\u8ba1\u65e5\u5fd7<\/p>\n<\/li>\n

  • \n

    \u6210\u672c\u4f18\u5316：\u6839\u636e\u4f7f\u7528\u6a21\u5f0f\u9009\u62e9\u5408\u9002\u7684\u4e91\u8d44\u6e90<\/p>\n<\/li>\n

    \u8fd9\u4e2a\u7cfb\u7edf\u4e3a\u673a\u5668\u5b66\u4e60\u6a21\u578b\u7684\u751f\u4ea7\u90e8\u7f72\u63d0\u4f9b\u4e86\u5b8c\u6574\u7684\u89e3\u51b3\u65b9\u6848，\u53ef\u4ee5\u5e2e\u52a9\u56e2\u961f\u5feb\u901f\u3001\u53ef\u9760\u5730\u5c06\u6a21\u578b\u6295\u5165\u751f\u4ea7\u73af\u5883\u3002<\/p>\n","protected":false},"excerpt":{"rendered":"

    \u4e00\u3001\u673a\u5668\u5b66\u4e60\u5728\u7ebf\u670d\u52a1\u6838\u5fc3\u67b6\u6784
    \n1.1 \u7edf\u4e00\u6a21\u578b\u670d\u52a1\u67b6\u6784
    \npython \u590d\u5236 \u4e0b\u8f7d
    \n\\”\\”\\”
    \n\u673a\u5668\u5b66\u4e60\u6a21\u578b\u5728\u7ebf\u670d\u52a1\u67b6\u6784
    \n\\”\\”\\”
    \nfrom abc import ABC, abstractmethod
    \nfrom typing import Dict, Any, List, Optional, Union
    \nfrom dataclasses import dataclass, asdict
    \nimport json
    \nimport time
    \nimport asyncio
    \nimport<\/p>\n","protected":false},"author":2,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[99,98,1375,292,190,100,207],"topic":[],"class_list":["post-65786","post","type-post","status-publish","format-standard","hentry","category-server","tag-java","tag-spring-boot","tag-word","tag-292","tag-190","tag-100","tag-207"],"yoast_head":"\n\u7c73\u54c8\u6e38Java\u9762\u8bd5\u88ab\u95ee\uff1a\u673a\u5668\u5b66\u4e60\u6a21\u578b\u7684\u5728\u7ebf\u670d\u52a1\u548cA\/B\u6d4b\u8bd5 - \u7f51\u7855\u4e92\u8054\u5e2e\u52a9\u4e2d\u5fc3<\/title>\n<meta name=\"robots\" content=\"index, follow, max-snippet:-1, max-image-preview:large, max-video-preview:-1\" \/>\n<link rel=\"canonical\" href=\"https:\/\/www.wsisp.com\/helps\/65786.html\" \/>\n<meta property=\"og:locale\" content=\"zh_CN\" \/>\n<meta property=\"og:type\" content=\"article\" \/>\n<meta property=\"og:title\" content=\"\u7c73\u54c8\u6e38Java\u9762\u8bd5\u88ab\u95ee\uff1a\u673a\u5668\u5b66\u4e60\u6a21\u578b\u7684\u5728\u7ebf\u670d\u52a1\u548cA\/B\u6d4b\u8bd5 - \u7f51\u7855\u4e92\u8054\u5e2e\u52a9\u4e2d\u5fc3\" \/>\n<meta property=\"og:description\" content=\"\u4e00\u3001\u673a\u5668\u5b66\u4e60\u5728\u7ebf\u670d\u52a1\u6838\u5fc3\u67b6\u6784 1.1 \u7edf\u4e00\u6a21\u578b\u670d\u52a1\u67b6\u6784 python \u590d\u5236 \u4e0b\u8f7d """ \u673a\u5668\u5b66\u4e60\u6a21\u578b\u5728\u7ebf\u670d\u52a1\u67b6\u6784 """ from abc import ABC, abstractmethod from typing import Dict, Any, List, Optional, Union from dataclasses import dataclass, asdict import json import time import asyncio import\" \/>\n<meta property=\"og:url\" content=\"https:\/\/www.wsisp.com\/helps\/65786.html\" \/>\n<meta property=\"og:site_name\" content=\"\u7f51\u7855\u4e92\u8054\u5e2e\u52a9\u4e2d\u5fc3\" \/>\n<meta property=\"article:published_time\" content=\"2026-01-25T09:40:35+00:00\" \/>\n<meta name=\"author\" content=\"admin\" \/>\n<meta name=\"twitter:card\" content=\"summary_large_image\" \/>\n<meta name=\"twitter:label1\" content=\"\u4f5c\u8005\" \/>\n\t<meta name=\"twitter:data1\" content=\"admin\" \/>\n\t<meta name=\"twitter:label2\" content=\"\u9884\u8ba1\u9605\u8bfb\u65f6\u95f4\" \/>\n\t<meta name=\"twitter:data2\" content=\"84 \u5206\" \/>\n<script type=\"application\/ld+json\" class=\"yoast-schema-graph\">{\"@context\":\"https:\/\/schema.org\",\"@graph\":[{\"@type\":\"WebPage\",\"@id\":\"https:\/\/www.wsisp.com\/helps\/65786.html\",\"url\":\"https:\/\/www.wsisp.com\/helps\/65786.html\",\"name\":\"\u7c73\u54c8\u6e38Java\u9762\u8bd5\u88ab\u95ee\uff1a\u673a\u5668\u5b66\u4e60\u6a21\u578b\u7684\u5728\u7ebf\u670d\u52a1\u548cA\/B\u6d4b\u8bd5 - \u7f51\u7855\u4e92\u8054\u5e2e\u52a9\u4e2d\u5fc3\",\"isPartOf\":{\"@id\":\"https:\/\/www.wsisp.com\/helps\/#website\"},\"datePublished\":\"2026-01-25T09:40:35+00:00\",\"dateModified\":\"2026-01-25T09:40:35+00:00\",\"author\":{\"@id\":\"https:\/\/www.wsisp.com\/helps\/#\/schema\/person\/358e386c577a3ab51c4493330a20ad41\"},\"breadcrumb\":{\"@id\":\"https:\/\/www.wsisp.com\/helps\/65786.html#breadcrumb\"},\"inLanguage\":\"zh-Hans\",\"potentialAction\":[{\"@type\":\"ReadAction\",\"target\":[\"https:\/\/www.wsisp.com\/helps\/65786.html\"]}]},{\"@type\":\"BreadcrumbList\",\"@id\":\"https:\/\/www.wsisp.com\/helps\/65786.html#breadcrumb\",\"itemListElement\":[{\"@type\":\"ListItem\",\"position\":1,\"name\":\"\u9996\u9875\",\"item\":\"https:\/\/www.wsisp.com\/helps\"},{\"@type\":\"ListItem\",\"position\":2,\"name\":\"\u7c73\u54c8\u6e38Java\u9762\u8bd5\u88ab\u95ee\uff1a\u673a\u5668\u5b66\u4e60\u6a21\u578b\u7684\u5728\u7ebf\u670d\u52a1\u548cA\/B\u6d4b\u8bd5\"}]},{\"@type\":\"WebSite\",\"@id\":\"https:\/\/www.wsisp.com\/helps\/#website\",\"url\":\"https:\/\/www.wsisp.com\/helps\/\",\"name\":\"\u7f51\u7855\u4e92\u8054\u5e2e\u52a9\u4e2d\u5fc3\",\"description\":\"\u9999\u6e2f\u670d\u52a1\u5668_\u9999\u6e2f\u4e91\u670d\u52a1\u5668\u8d44\u8baf_\u670d\u52a1\u5668\u5e2e\u52a9\u6587\u6863_\u670d\u52a1\u5668\u6559\u7a0b\",\"potentialAction\":[{\"@type\":\"SearchAction\",\"target\":{\"@type\":\"EntryPoint\",\"urlTemplate\":\"https:\/\/www.wsisp.com\/helps\/?s={search_term_string}\"},\"query-input\":\"required name=search_term_string\"}],\"inLanguage\":\"zh-Hans\"},{\"@type\":\"Person\",\"@id\":\"https:\/\/www.wsisp.com\/helps\/#\/schema\/person\/358e386c577a3ab51c4493330a20ad41\",\"name\":\"admin\",\"image\":{\"@type\":\"ImageObject\",\"inLanguage\":\"zh-Hans\",\"@id\":\"https:\/\/www.wsisp.com\/helps\/#\/schema\/person\/image\/\",\"url\":\"https:\/\/gravatar.wp-china-yes.net\/avatar\/?s=96&d=mystery\",\"contentUrl\":\"https:\/\/gravatar.wp-china-yes.net\/avatar\/?s=96&d=mystery\",\"caption\":\"admin\"},\"sameAs\":[\"http:\/\/wp.wsisp.com\"],\"url\":\"https:\/\/www.wsisp.com\/helps\/author\/admin\"}]}<\/script>\n<!-- \/ Yoast SEO plugin. -->","yoast_head_json":{"title":"\u7c73\u54c8\u6e38Java\u9762\u8bd5\u88ab\u95ee\uff1a\u673a\u5668\u5b66\u4e60\u6a21\u578b\u7684\u5728\u7ebf\u670d\u52a1\u548cA\/B\u6d4b\u8bd5 - \u7f51\u7855\u4e92\u8054\u5e2e\u52a9\u4e2d\u5fc3","robots":{"index":"index","follow":"follow","max-snippet":"max-snippet:-1","max-image-preview":"max-image-preview:large","max-video-preview":"max-video-preview:-1"},"canonical":"https:\/\/www.wsisp.com\/helps\/65786.html","og_locale":"zh_CN","og_type":"article","og_title":"\u7c73\u54c8\u6e38Java\u9762\u8bd5\u88ab\u95ee\uff1a\u673a\u5668\u5b66\u4e60\u6a21\u578b\u7684\u5728\u7ebf\u670d\u52a1\u548cA\/B\u6d4b\u8bd5 - \u7f51\u7855\u4e92\u8054\u5e2e\u52a9\u4e2d\u5fc3","og_description":"\u4e00\u3001\u673a\u5668\u5b66\u4e60\u5728\u7ebf\u670d\u52a1\u6838\u5fc3\u67b6\u6784 1.1 \u7edf\u4e00\u6a21\u578b\u670d\u52a1\u67b6\u6784 python \u590d\u5236 \u4e0b\u8f7d \"\"\" \u673a\u5668\u5b66\u4e60\u6a21\u578b\u5728\u7ebf\u670d\u52a1\u67b6\u6784 \"\"\" from abc import ABC, abstractmethod from typing import Dict, Any, List, Optional, Union from dataclasses import dataclass, asdict import json import time import asyncio import","og_url":"https:\/\/www.wsisp.com\/helps\/65786.html","og_site_name":"\u7f51\u7855\u4e92\u8054\u5e2e\u52a9\u4e2d\u5fc3","article_published_time":"2026-01-25T09:40:35+00:00","author":"admin","twitter_card":"summary_large_image","twitter_misc":{"\u4f5c\u8005":"admin","\u9884\u8ba1\u9605\u8bfb\u65f6\u95f4":"84 \u5206"},"schema":{"@context":"https:\/\/schema.org","@graph":[{"@type":"WebPage","@id":"https:\/\/www.wsisp.com\/helps\/65786.html","url":"https:\/\/www.wsisp.com\/helps\/65786.html","name":"\u7c73\u54c8\u6e38Java\u9762\u8bd5\u88ab\u95ee\uff1a\u673a\u5668\u5b66\u4e60\u6a21\u578b\u7684\u5728\u7ebf\u670d\u52a1\u548cA\/B\u6d4b\u8bd5 - \u7f51\u7855\u4e92\u8054\u5e2e\u52a9\u4e2d\u5fc3","isPartOf":{"@id":"https:\/\/www.wsisp.com\/helps\/#website"},"datePublished":"2026-01-25T09:40:35+00:00","dateModified":"2026-01-25T09:40:35+00:00","author":{"@id":"https:\/\/www.wsisp.com\/helps\/#\/schema\/person\/358e386c577a3ab51c4493330a20ad41"},"breadcrumb":{"@id":"https:\/\/www.wsisp.com\/helps\/65786.html#breadcrumb"},"inLanguage":"zh-Hans","potentialAction":[{"@type":"ReadAction","target":["https:\/\/www.wsisp.com\/helps\/65786.html"]}]},{"@type":"BreadcrumbList","@id":"https:\/\/www.wsisp.com\/helps\/65786.html#breadcrumb","itemListElement":[{"@type":"ListItem","position":1,"name":"\u9996\u9875","item":"https:\/\/www.wsisp.com\/helps"},{"@type":"ListItem","position":2,"name":"\u7c73\u54c8\u6e38Java\u9762\u8bd5\u88ab\u95ee\uff1a\u673a\u5668\u5b66\u4e60\u6a21\u578b\u7684\u5728\u7ebf\u670d\u52a1\u548cA\/B\u6d4b\u8bd5"}]},{"@type":"WebSite","@id":"https:\/\/www.wsisp.com\/helps\/#website","url":"https:\/\/www.wsisp.com\/helps\/","name":"\u7f51\u7855\u4e92\u8054\u5e2e\u52a9\u4e2d\u5fc3","description":"\u9999\u6e2f\u670d\u52a1\u5668_\u9999\u6e2f\u4e91\u670d\u52a1\u5668\u8d44\u8baf_\u670d\u52a1\u5668\u5e2e\u52a9\u6587\u6863_\u670d\u52a1\u5668\u6559\u7a0b","potentialAction":[{"@type":"SearchAction","target":{"@type":"EntryPoint","urlTemplate":"https:\/\/www.wsisp.com\/helps\/?s={search_term_string}"},"query-input":"required name=search_term_string"}],"inLanguage":"zh-Hans"},{"@type":"Person","@id":"https:\/\/www.wsisp.com\/helps\/#\/schema\/person\/358e386c577a3ab51c4493330a20ad41","name":"admin","image":{"@type":"ImageObject","inLanguage":"zh-Hans","@id":"https:\/\/www.wsisp.com\/helps\/#\/schema\/person\/image\/","url":"https:\/\/gravatar.wp-china-yes.net\/avatar\/?s=96&d=mystery","contentUrl":"https:\/\/gravatar.wp-china-yes.net\/avatar\/?s=96&d=mystery","caption":"admin"},"sameAs":["http:\/\/wp.wsisp.com"],"url":"https:\/\/www.wsisp.com\/helps\/author\/admin"}]}},"_links":{"self":[{"href":"https:\/\/www.wsisp.com\/helps\/wp-json\/wp\/v2\/posts\/65786","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.wsisp.com\/helps\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.wsisp.com\/helps\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.wsisp.com\/helps\/wp-json\/wp\/v2\/users\/2"}],"replies":[{"embeddable":true,"href":"https:\/\/www.wsisp.com\/helps\/wp-json\/wp\/v2\/comments?post=65786"}],"version-history":[{"count":0,"href":"https:\/\/www.wsisp.com\/helps\/wp-json\/wp\/v2\/posts\/65786\/revisions"}],"wp:attachment":[{"href":"https:\/\/www.wsisp.com\/helps\/wp-json\/wp\/v2\/media?parent=65786"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.wsisp.com\/helps\/wp-json\/wp\/v2\/categories?post=65786"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.wsisp.com\/helps\/wp-json\/wp\/v2\/tags?post=65786"},{"taxonomy":"topic","embeddable":true,"href":"https:\/\/www.wsisp.com\/helps\/wp-json\/wp\/v2\/topic?post=65786"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}