MD5 - 计算 MD5 哈希值

# 计算字符串的 MD5
hash_value = MD5("Hello World")

# 计算字节数据的 MD5
hash_value = MD5(b"Hello World")

# 计算字符串的 MD5
hash1 = MD5("Hello World")
Log(f"MD5: {hash1}")
# 输出: b10a8db164e0754105b7a99be72e3fe5

# 相同内容产生相同哈希
hash2 = MD5("Hello World")
Log(f"哈希相等: {hash1 == hash2}")  # True

import time

def generate_signature(api_key, secret_key, params):
    """生成 API 请求签名"""
    # 按键名排序参数
    sorted_params = sorted(params.items())
    
    # 拼接参数字符串
    param_str = '&'.join([f"{k}={v}" for k, v in sorted_params])
    
    # 添加密钥
    sign_str = f"{api_key}{param_str}{secret_key}"
    
    # 计算 MD5 签名
    signature = MD5(sign_str)
    
    return signature

# 使用示例
api_key = "your_api_key"
secret_key = "your_secret_key"
params = {
    "symbol": "BTC_USDT",
    "side": "buy",
    "amount": "0.1",
    "timestamp": str(int(time.time()))
}

signature = generate_signature(api_key, secret_key, params)
Log(f"请求签名: {signature}")

def save_data_with_checksum(key, data):
    """保存数据并记录校验和"""
    import json
    
    # 序列化数据
    data_str = json.dumps(data)
    
    # 计算 MD5 校验和
    checksum = MD5(data_str)
    
    # 保存数据和校验和
    _G(key, data)
    _G(f"{key}_checksum", checksum)
    
    Log(f"数据已保存，校验和: {checksum}")
    return checksum

def load_data_with_verification(key):
    """加载数据并验证完整性"""
    import json
    
    # 加载数据和校验和
    data = _G(key)
    saved_checksum = _G(f"{key}_checksum")
    
    if data is None:
        Log("数据不存在")
        return None
    
    # 计算当前数据的校验和
    data_str = json.dumps(data)
    current_checksum = MD5(data_str)
    
    # 验证完整性
    if current_checksum == saved_checksum:
        Log("✅ 数据完整性验证通过")
        return data
    else:
        Log("❌ 数据可能已损坏！", level="ERROR")
        return None

# 使用示例
config = {"risk_level": 0.5, "max_position": 10}
save_data_with_checksum("strategy_config", config)

# 稍后加载
loaded_config = load_data_with_verification("strategy_config")
Log(f"配置: {loaded_config}")

processed_orders = set()

def process_order(order):
    """处理订单，避免重复处理"""
    import json
    
    # 使用订单数据的 MD5 作为唯一标识
    order_hash = MD5(json.dumps(order, sort_keys=True))
    
    if order_hash in processed_orders:
        Log(f"订单已处理过，跳过: {order['Id']}")
        return False
    
    # 处理订单
    Log(f"处理新订单: {order['Id']}")
    # ... 订单处理逻辑 ...
    
    # 标记为已处理
    processed_orders.add(order_hash)
    return True

# 使用示例
orders = exchange.GetOrders()
for order in orders:
    process_order(order)

def get_cached_data(symbol, period, use_cache=True):
    """获取K线数据（带缓存）"""
    import json
    
    # 生成缓存键
    cache_key_data = {"symbol": symbol, "period": period}
    cache_key = f"kline_{MD5(json.dumps(cache_key_data, sort_keys=True))}"
    
    # 尝试从缓存获取
    if use_cache:
        cached = _G(cache_key)
        if cached:
            Log(f"✅ 使用缓存数据: {symbol} {period}")
            return cached
    
    # 获取新数据
    Log(f"📡 请求新数据: {symbol} {period}")
    exchange.SetCurrency(symbol)
    records = exchange.GetRecords(period)
    
    # 保存到缓存
    _G(cache_key, records)
    
    return records

# 使用示例
records1 = get_cached_data("BTC_USDT", "1h")  # 请求新数据
records2 = get_cached_data("BTC_USDT", "1h")  # 使用缓存

def hash_password(password, salt=""):
    """对密码进行哈希（加盐）"""
    # 注意: MD5 不够安全，实际应用应使用更安全的算法
    salted = f"{salt}{password}{salt}"
    return MD5(salted)

def verify_password(password, hashed, salt=""):
    """验证密码"""
    return hash_password(password, salt) == hashed

# 使用示例（仅用于演示，实际请使用更安全的方法）
password = "MySecretPassword123"
salt = UUID()  # 使用随机盐

hashed = hash_password(password, salt)
Log(f"密码哈希: {hashed}")
Log(f"盐值: {salt}")

# 验证
is_valid = verify_password("MySecretPassword123", hashed, salt)
Log(f"密码验证: {is_valid}")  # True

is_valid = verify_password("WrongPassword", hashed, salt)
Log(f"错误密码: {is_valid}")  # False

def calculate_strategy_fingerprint():
    """计算策略代码的指纹"""
    try:
        with open(__file__, 'r', encoding='utf-8') as f:
            code = f.read()
        
        fingerprint = MD5(code)
        Log(f"策略代码指纹: {fingerprint}")
        
        # 保存指纹，用于检测代码是否被修改
        last_fingerprint = _G("strategy_fingerprint")
        if last_fingerprint and last_fingerprint != fingerprint:
            Log("⚠️  检测到策略代码已修改！", level="WARNING")
        
        _G("strategy_fingerprint", fingerprint)
        return fingerprint
    except Exception as e:
        Log(f"计算指纹失败: {e}")
        return None

# 策略启动时检查
calculate_strategy_fingerprint()