技术研发注意事项

缓存决策 = f(数据变化频率, 用户对新鲜度的敏感度, 计算成本, 存储成本)

12:00  工程师 push 代码
12:01  实时 logging 检测到异常（错误率从 0.1% 飙升到 5%）
12:02  告警触发，SRE 介入
12:03  确认 push 时间与告警时间一致
12:04  Revert 代码，服务恢复

import logging
import json
from datetime import datetime, timezone

# 结构化日志格式
logger = logging.getLogger('trading')
logger.setLevel(logging.INFO)

# 日志格式：JSON 便于后续分析
class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            'timestamp': datetime.now(timezone.utc).isoformat(),  # utcnow() 自 3.12 弃用
            'level': record.levelname,
            'module': record.name,
            'message': record.getMessage(),
        }
        if hasattr(record, 'extra_fields'):
            log_entry.update(record.extra_fields)
        return json.dumps(log_entry)

# 日志记录示例
def log_order(order_id, symbol, side, price, amount, status):
    """记录订单事件的结构化日志"""
    logger.info(f"Order {order_id}: {side} {amount} {symbol} @ {price}",
                extra={
                    'extra_fields': {
                        'event_type': 'order',
                        'order_id': order_id,
                        'symbol': symbol,
                        'side': side,
                        'price': price,
                        'amount': amount,
                        'status': status
                    }
                })

import time
import functools
from collections import defaultdict

# 简单的延迟统计装饰器
class LatencyStats:
    def __init__(self):
        self.stats = defaultdict(list)

    def record(self, name, latency_ms):
        self.stats[name].append(latency_ms)

    def report(self):
        for name, latencies in self.stats.items():
            if latencies:
                avg = sum(latencies) / len(latencies)
                p99 = sorted(latencies)[int(len(latencies) * 0.99)]
                print(f"{name}: avg={avg:.2f}ms, p99={p99:.2f}ms")

stats = LatencyStats()

def profile(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        latency = (time.perf_counter() - start) * 1000
        stats.record(func.__name__, latency)
        return result
    return wrapper

@profile
def search(query):
    # 模拟搜索操作
    time.sleep(0.05)
    return ["result1", "result2"]

@profile
def rank(results):
    # 模拟排序操作
    time.sleep(0.02)
    return sorted(results)