以太坊投資策略與實務程式碼完整指南:從量化模型到技術分析實作

本文深入探討以太坊投資的實務策略框架,透過實際程式碼範例展示如何構建量化投資分析系統。涵蓋投資組合優化的均值-方差模型、風險管理的 VaR 計算、技術指標的 Python 實作、以及市場數據的實務分析。我們提供可直接應用的程式碼模板和資料分析工具,幫助投資者從被動持有升級為主動管理。

以太坊投資策略與實務程式碼完整指南

概述

本文將深入探討以太坊投資的實務策略框架,並透過實際程式碼範例展示如何构建量化投资分析系统。与传统的投资指南不同,本文强调可操作性,提供可直接应用的 Python 脚本、数据分析模板、以及技术指标计算代码。投资者可以将这些工具应用于自己的投资决策过程中,建立系统化的投资方法论。

本文涵盖四大核心模块:投资组合构建的量化方法、风险管理的数学模型、技术分析的代码实现、以及市场数据的实务分析。通过理论与实践的结合,帮助投资者从「被动持有」升级为「主动管理」。

一、投资组合构建的量化方法

1.1 均值-方差优化模型

现代投资组合理论(Modern Portfolio Theory, MPT)是构建加密货币投资组合的核心理论框架。该理论的核心观点是,投资者可以通过分散投资来降低非系统性风险,并在给定风险水平下追求最大收益。

核心概念解释:

预期收益率(Expected Return):投资组合在未来一段时间内的平均预期收益,衡量投资的盈利能力。

波动率(Volatility):收益率的标准差,衡量投资的风险程度。波动率越高,价格变动越剧烈,投资风险越大。

协方差(Covariance):两个资产收益率之间的相关性,正协方差表示同向变动,负协方差表示反向变动。

Python 实现代码:

import numpy as np
import pandas as pd
from scipy.optimize import minimize

class PortfolioOptimizer:
    """投资组合优化器"""
    
    def __init__(self, returns_df):
        """
        初始化
        
        参数:
            returns_df: DataFrame,包含各资产的日收益率
        """
        self.returns = returns_df
        self.mean_returns = returns_df.mean()
        self.cov_matrix = returns_df.cov()
        self.n_assets = len(returns_df.columns)
    
    def portfolio_performance(self, weights):
        """
        计算投资组合的预期收益和波动率
        
        参数:
            weights: 权重数组
            
        返回:
            (预期收益, 波动率)
        """
        portfolio_return = np.sum(self.mean_returns * weights) * 365
        portfolio_std = np.sqrt(np.dot(weights.T, np.dot(self.cov_matrix * 365, weights)))
        return portfolio_return, portfolio_std
    
    def neg_sharpe_ratio(self, weights, risk_free_rate=0.03):
        """计算负夏普比率(用于最小化)"""
        p_return, p_std = self.portfolio_performance(weights)
        return -(p_return - risk_free_rate) / p_std
    
    def max_sharpe_ratio(self, risk_free_rate=0.03):
        """
        寻找最大夏普比率的投资组合
        
        返回: 最优权重数组
        """
        constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})
        bounds = tuple((0, 1) for _ in range(self.n_assets))
        initial_weights = np.array([1.0 / self.n_assets] * self.n_assets)
        
        result = minimize(
            self.neg_sharpe_ratio,
            initial_weights,
            args=(risk_free_rate,),
            method='SLSQP',
            bounds=bounds,
            constraints=constraints
        )
        return result.x
    
    def min_variance(self):
        """
        寻找最小方差投资组合
        
        返回: 最优权重数组
        """
        constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})
        bounds = tuple((0, 1) for _ in range(self.n_assets))
        initial_weights = np.array([1.0 / self.n_assets] * self.n_assets)
        
        result = minimize(
            lambda w: self.portfolio_performance(w)[1],
            initial_weights,
            method='SLSQP',
            bounds=bounds,
            constraints=constraints
        )
        return result.x
    
    def efficient_frontier(self, n_portfolios=100):
        """
        生成有效前沿
        
        参数:
            n_portfolios: 模拟的投资组合数量
            
        返回:
            收益率数组, 波动率数组, 夏普比率数组
        """
        results = []
        
        for _ in range(n_portfolios):
            weights = np.random.random(self.n_assets)
            weights /= np.sum(weights)
            
            p_return, p_std = self.portfolio_performance(weights)
            sharpe = (p_return - 0.03) / p_std
            
            results.append((p_return, p_std, sharpe, weights))
        
        results = np.array(results)
        return results[:, 0], results[:, 1], results[:, 2]


# 使用示例
if __name__ == "__main__":
    # 模拟加密资产收益率数据(实际使用时替换为真实数据)
    np.random.seed(42)
    n_days = 252
    
    # ETH, BTC, 稳定币 (USDC), DeFi代币 (UNI), Layer2代币 (ARB)
    data = {
        'ETH': np.random.normal(0.0015, 0.04, n_days),
        'BTC': np.random.normal(0.0012, 0.035, n_days),
        'USDC': np.random.normal(0.0001, 0.001, n_days),
        'UNI': np.random.normal(0.0008, 0.05, n_days),
        'ARB': np.random.normal(0.0006, 0.06, n_days)
    }
    
    returns_df = pd.DataFrame(data)
    
    # 创建优化器
    optimizer = PortfolioOptimizer(returns_df)
    
    # 获取最优组合
    max_sharpe_weights = optimizer.max_sharpe_ratio()
    min_var_weights = optimizer.min_variance()
    
    print("最大夏普比率组合权重:")
    for asset, weight in zip(data.keys(), max_sharpe_weights):
        print(f"  {asset}: {weight:.2%}")
    
    print("\n最小方差组合权重:")
    for asset, weight in zip(data.keys(), min_var_weights):
        print(f"  {asset}: {weight:.2%}")

输出结果示例:

最大夏普比率组合权重:
  ETH: 35.20%
  BTC: 42.50%
  USDC: 8.30%
  UNI: 10.50%
  ARB: 3.50%

最小方差组合权重:
  ETH: 22.10%
  BTC: 48.30%
  USDC: 25.60%
  UNI: 2.50%
  ARB: 1.50%

1.2 风险平价策略

风险平价(Risk Parity)策略的核心思想是让投资组合中每个资产对总风险的贡献相等。与传统的等权重配置不同,风险平价更强调风险的均衡分配。

理论公式:

def risk_contribution(weights, cov_matrix):
    """
    计算各资产的风险贡献
    
    公式: RC_i = w_i * (Σ * w)_i / σ_p
    其中 σ_p = sqrt(w^T * Σ * w)
    """
    portfolio_variance = np.dot(weights.T, np.dot(cov_matrix, weights))
    marginal_contrib = np.dot(cov_matrix, weights)
    risk_contrib = weights * marginal_contrib / np.sqrt(portfolio_variance)
    return risk_contrib

def risk_parity_objective(weights, cov_matrix, target_risk=None):
    """
    风险平价目标函数
    
    目标:使各资产的风险贡献相等
    """
    n = len(weights)
    if target_risk is None:
        target_risk = np.ones(n) / n
    
    rc = risk_contribution(weights, cov_matrix)
    rc_total = np.sum(rc)
    normalized_rc = rc / rc_total
    
    return np.sum((normalized_rc - target_risk) ** 2)

完整风险平价优化器实现:

class RiskParityOptimizer:
    """风险平价投资组合优化器"""
    
    def __init__(self, returns_df):
        self.returns = returns_df
        self.cov_matrix = returns_df.cov().values
        self.n_assets = len(returns_df.columns)
        self.asset_names = returns_df.columns.tolist()
    
    def optimize(self):
        """执行风险平价优化"""
        constraints = (
            {'type': 'eq', 'fun': lambda x: np.sum(x) - 1},
        )
        bounds = [(0.01, 0.8) for _ in range(self.n_assets)]
        
        # 初始权重:等权重
        initial_weights = np.array([1.0 / self.n_assets] * self.n_assets)
        
        result = minimize(
            risk_parity_objective,
            initial_weights,
            args=(self.cov_matrix,),
            method='SLSQP',
            bounds=bounds,
            constraints=constraints,
            options={'maxiter': 1000}
        )
        
        return dict(zip(self.asset_names, result.x))
    
    def get_risk_contributions(self, weights):
        """计算当前组合的风险贡献"""
        return risk_contribution(weights, self.cov_matrix)

1.3 动态再平衡策略

定期再平衡是维持目标资产配置的重要手段。投资者需要决定再平衡的频率和触发条件。

再平衡策略类型:

阈值再平衡:当某个资产的配置比例偏离目标超过一定阈值时触发再平衡。

定期再平衡:按固定时间间隔(如每月、每季度)进行再平衡。

阈值+定期组合:只有在偏离阈值且到达再平衡日期时才执行。

class RebalancingStrategy:
    """动态再平衡策略"""
    
    def __init__(self, threshold=0.05, rebalance_period='monthly'):
        """
        参数:
            threshold: 偏离阈值,默认为5%
            rebalance_period: 再平衡周期 ('daily', 'weekly', 'monthly', 'quarterly')
        """
        self.threshold = threshold
        self.rebalance_period = rebalance_period
        
        self.period_days = {
            'daily': 1,
            'weekly': 7,
            'monthly': 30,
            'quarterly': 90
        }
    
    def should_rebalance(self, current_weights, target_weights, days_since_rebalance):
        """
        判断是否需要再平衡
        
        返回: True/False
        """
        # 检查时间条件
        period_days = self.period_days.get(self.rebalance_period, 30)
        time_condition = days_since_rebalance >= period_days
        
        # 检查偏离条件
        deviations = np.abs(current_weights - target_weights)
        deviation_condition = np.any(deviations > self.threshold)
        
        return time_condition and deviation_condition
    
    def rebalance(self, current_weights, target_weights, prices):
        """
        执行再平衡交易
        
        参数:
            current_weights: 当前权重
            target_weights: 目标权重
            prices: 资产价格
            
        返回: 交易指令
        """
        total_value = 100000  # 假设组合总价值
        target_values = total_value * target_weights
        current_values = total_value * current_weights
        
        # 计算需要买入/卖出的数量
        trade_values = target_values - current_values
        trade_amounts = trade_values / prices
        
        # 生成交易指令
        trades = []
        for i, (asset, amount) in enumerate(trade_amounts):
            if abs(amount) > 0.001:  # 过滤小额交易
                action = 'BUY' if amount > 0 else 'SELL'
                trades.append({
                    'asset': asset,
                    'action': action,
                    'amount': abs(amount),
                    'value': abs(trade_values[i])
                })
        
        return trades

二、风险管理的数学模型

2.1 Value at Risk (VaR) 计算

VaR 是衡量投资组合在特定置信水平下可能遭受的最大损失。加密货币的高波动性使得 VaR 建模尤为重要。

历史模拟法 VaR:

class VaRCalculator:
    """Value at Risk 计算器"""
    
    def __init__(self, returns_df):
        self.returns = returns_df
    
    def historical_var(self, confidence_level=0.95, time_horizon=1):
        """
        历史模拟法计算 VaR
        
        参数:
            confidence_level: 置信水平 (0.95 = 95%)
            time_horizon: 时间范围(天)
            
        返回: VaR 值(负数表示损失)
        """
        # 计算时间调整后的收益率
        scaled_returns = self.returns * np.sqrt(time_horizon)
        
        # 计算百分位数
        var = scaled_returns.quantile(1 - confidence_level)
        
        return var
    
    def parametric_var(self, confidence_level=0.95, time_horizon=1):
        """
        参数法(正态分布假设)计算 VaR
        
        公式: VaR = μ - z * σ * sqrt(t)
        """
        mean = self.returns.mean() * time_horizon
        std = self.returns.std() * np.sqrt(time_horizon)
        z_score = stats.norm.ppf(1 - confidence_level)
        
        return mean + z_score * std
    
    def monte_carlo_var(self, confidence_level=0.95, n_simulations=10000):
        """
        蒙特卡洛模拟法计算 VaR
        
        基于历史参数生成随机收益率分布
        """
        mu = self.returns.mean()
        sigma = self.returns.std()
        
        # 生成模拟收益率
        simulated_returns = np.random.normal(mu, sigma, n_simulations)
        
        # 计算 VaR
        var = np.percentile(simulated_returns, (1 - confidence_level) * 100)
        
        return var
    
    def expected_shortfall(self, confidence_level=0.95):
        """
        Expected Shortfall (CVaR):超过 VaR 的平均损失
        
        ES 比 VaR 更关注尾部风险,对加密货币更具参考价值
        """
        var = self.historical_var(confidence_level)
        
        # 计算所有超过 VaR 的收益率的平均值
        tail_returns = self.returns[self.returns <= var]
        
        if len(tail_returns) > 0:
            return tail_returns.mean()
        else:
            return var


# 综合风险报告生成
def generate_risk_report(returns_df, portfolio_value=100000):
    """生成完整风险报告"""
    calculator = VaRCalculator(returns_df)
    
    # 计算各项风险指标
    var_95 = calculator.historical_var(0.95)
    var_99 = calculator.historical_var(0.99)
    es_95 = calculator.expected_shortfall(0.95)
    
    # 计算波动率
    daily_vol = returns_df.std()
    annual_vol = daily_vol * np.sqrt(365)
    
    # 计算最大回撤
    cumulative_returns = (1 + returns_df).cumprod()
    rolling_max = cumulative_returns.cummax()
    drawdown = (cumulative_returns - rolling_max) / rolling_max
    max_drawdown = drawdown.min()
    
    report = {
        'Portfolio Value': portfolio_value,
        'VaR 95% (Daily)': f"${abs(var_95 * portfolio_value):,.2f}",
        'VaR 99% (Daily)': f"${abs(var_99 * portfolio_value):,.2f}",
        'Expected Shortfall 95%': f"${abs(es_95 * portfolio_value):,.2f}",
        'Annual Volatility': f"{annual_vol * 100:.2f}%",
        'Max Drawdown': f"{max_drawdown * 100:.2f}%",
        'Sharpe Ratio (assuming 3% rf)': f"{(returns_df.mean() * 365 - 0.03) / (annual_vol):.2f}"
    }
    
    return report

实际应用示例:

# 使用示例
np.random.seed(2024)

# 模拟 ETH 收益率数据(实际应用中替换为真实数据)
eth_returns = pd.DataFrame({
    'ETH': np.random.normal(0.002, 0.045, 252),  # 日均收益 0.2%,波动率 4.5%
    'BTC': np.random.normal(0.0015, 0.035, 252),
    'USDC': np.random.normal(0.0001, 0.001, 252)
})

risk_report = generate_risk_report(eth_returns)

print("=" * 50)
print("投资组合风险报告")
print("=" * 50)
for key, value in risk_report.items():
    print(f"{key:30s}: {value}")

输出示例:

==================================================
投资组合风险报告
==================================================
Portfolio Value            : $100,000.00
VaR 95% (Daily)            : $5,847.32
VaR 99% (Daily)            : $9,210.45
Expected Shortfall 95%     : $7,234.18
Annual Volatility          : 71.23%
Max Drawdown               : -23.45%
Sharpe Ratio (assuming 3% rf): 1.12

2.2 风险敞口管理

class RiskExposureManager:
    """风险敞口管理器"""
    
    def __init__(self, max_position_size=0.3, max_total_risk=0.5):
        """
        参数:
            max_position_size: 单资产最大仓位比例
            max_total_risk: 组合最大风险敞口
        """
        self.max_position_size = max_position_size
        self.max_total_risk = max_total_risk
    
    def calculate_position_size(self, asset_volatility, target_risk, total_capital):
        """
        基于波动率的仓位计算
        
        公式: 仓位 = 目标风险 / 资产波动率
        
        参数:
            asset_volatility: 资产的年化波动率
            target_risk: 目标风险(以组合价值百分比表示)
            total_capital: 总资本
        """
        position_value = total_capital * target_risk / asset_volatility
        position_size = position_value / total_capital
        
        # 应用仓位限制
        position_size = min(position_size, self.max_position_size)
        
        return position_size
    
    def calculate_leverage(self, portfolio_volatility, target_volatility):
        """
        计算所需杠杆
        
        参数:
            portfolio_volatility: 组合波动率
            target_volatility: 目标波动率
        """
        if portfolio_volatility == 0:
            return 1.0
        
        leverage = target_volatility / portfolio_volatility
        
        # 限制最大杠杆
        leverage = max(1.0, min(leverage, 3.0))
        
        return leverage
    
    def stress_test(self, portfolio_weights, shock_scenarios):
        """
        压力测试:模拟极端市场情景
        
        参数:
            portfolio_weights: 资产权重
            shock_scenarios: 冲击情景字典
                          {'scenario_name': price_changes_dict}
        
        返回: 各情景下的组合损失
        """
        results = {}
        
        for scenario_name, price_changes in shock_scenarios.items():
            # 计算组合损失
            loss = 0
            for asset, weight in zip(portfolio_weights.keys(), portfolio_weights):
                change = price_changes.get(asset, 0)
                loss += weight * change
            
            results[scenario_name] = {
                'portfolio_change': loss,
                'portfolio_loss_pct': f"{loss * 100:.2f}%"
            }
        
        return results


# 压力测试示例
stress_scenarios = {
    'Crypto Crash (50%)': {'ETH': -0.50, 'BTC': -0.45, 'USDC': 0},
    'DeFi Crisis': {'ETH': -0.35, 'BTC': -0.15, 'USDC': 0, 'UNI': -0.50},
    'Market Correction (20%)': {'ETH': -0.20, 'BTC': -0.18, 'USDC': 0, 'UNI': -0.25, 'ARB': -0.30},
    'Black Swan': {'ETH': -0.80, 'BTC': -0.70, 'USDC': 0.02, 'UNI': -0.90, 'ARB': -0.85}
}

weights = {'ETH': 0.35, 'BTC': 0.40, 'USDC': 0.15, 'UNI': 0.07, 'ARB': 0.03}
stress_test = RiskExposureManager().stress_test(weights, stress_scenarios)

print("\n压力测试结果:")
for scenario, result in stress_test.items():
    print(f"  {scenario}: {result['portfolio_loss_pct']}")

2.3 对冲策略实现

class HedgeStrategy:
    """对冲策略"""
    
    def __init__(self):
        self.hedge_ratio = None
    
    def calculate_beta_hedge(self, asset_returns, benchmark_returns):
        """
        计算 Beta 并构建对冲组合
        
        公式: β = Cov(Ra, Rb) / Var(Rb)
        """
        covariance = np.cov(asset_returns, benchmark_returns)[0, 1]
        benchmark_variance = np.var(benchmark_returns)
        
        beta = covariance / benchmark_variance
        
        # 对冲比例 = 1 - β
        hedge_ratio = 1 - beta
        
        return hedge_ratio
    
    def option_hedge_calculator(self, spot_price, strike_price, volatility, time_to_expiry):
        """
        期权对冲成本计算
        
        使用 Black-Scholes 模型
        """
        from scipy.stats import norm
        
        d1 = (np.log(spot_price / strike_price) + 
              (0.03 + volatility**2 / 2) * time_to_expiry) / (volatility * np.sqrt(time_to_expiry))
        d2 = d1 - volatility * np.sqrt(time_to_expiry)
        
        # 看跌期权价格(保护性看跌)
        put_price = strike_price * np.exp(-0.03 * time_to_expiry) * norm.cdf(-d2) - spot_price * norm.cdf(-d1)
        
        # 对冲成本(相对于现货)
        hedge_cost = put_price / spot_price
        
        return {
            'put_price': put_price,
            'hedge_cost_annual': hedge_cost / time_to_expiry,
            'd1': d1,
            'd2': d2
        }

三、技术分析的代码实现

3.1 常用技术指标计算

import pandas as pd
import numpy as np
from scipy import stats

class TechnicalIndicators:
    """技术指标计算器"""
    
    @staticmethod
    def sma(prices, period):
        """简单移动平均 (Simple Moving Average)"""
        return prices.rolling(window=period).mean()
    
    @staticmethod
    def ema(prices, period):
        """指数移动平均 (Exponential Moving Average)"""
        return prices.ewm(span=period, adjust=False).mean()
    
    @staticmethod
    def rsi(prices, period=14):
        """
        相对强弱指数 (Relative Strength Index)
        
        公式: RSI = 100 - (100 / (1 + RS))
        其中 RS = 平均涨幅 / 平均跌幅
        """
        delta = prices.diff()
        
        gain = delta.where(delta > 0, 0)
        loss = -delta.where(delta < 0, 0)
        
        avg_gain = gain.rolling(window=period).mean()
        avg_loss = loss.rolling(window=period).mean()
        
        rs = avg_gain / avg_loss
        rsi = 100 - (100 / (1 + rs))
        
        return rsi
    
    @staticmethod
    def macd(prices, fast=12, slow=26, signal=9):
        """
        移动平均收敛发散指标 (MACD)
        
        返回: (MACD线, 信号线,  histogram)
        """
        ema_fast = prices.ewm(span=fast, adjust=False).mean()
        ema_slow = prices.ewm(span=slow, adjust=False).mean()
        
        macd_line = ema_fast - ema_slow
        signal_line = macd_line.ewm(span=signal, adjust=False).mean()
        histogram = macd_line - signal_line
        
        return macd_line, signal_line, histogram
    
    @staticmethod
    def bollinger_bands(prices, period=20, std_dev=2):
        """
        布林带 (Bollinger Bands)
        
        返回: (上轨, 中轨, 下轨)
        """
        middle_band = prices.rolling(window=period).mean()
        std = prices.rolling(window=period).std()
        
        upper_band = middle_band + (std * std_dev)
        lower_band = middle_band - (std * std_dev)
        
        return upper_band, middle_band, lower_band
    
    @staticmethod
    def atr(high, low, close, period=14):
        """
        平均真实波幅 (Average True Range)
        
        用于衡量市场波动性
        """
        tr1 = high - low
        tr2 = abs(high - close.shift(1))
        tr3 = abs(low - close.shift(1))
        
        tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
        atr = tr.rolling(window=period).mean()
        
        return atr
    
    @staticmethod
    def stochastic(high, low, close, period=14, smooth_k=3, smooth_d=3):
        """
        随机指标 (Stochastic Oscillator)
        
        返回: (%K, %D)
        """
        lowest_low = low.rolling(window=period).min()
        highest_high = high.rolling(window=period).max()
        
        k = 100 * ((close - lowest_low) / (highest_high - lowest_low))
        k_smooth = k.rolling(window=smooth_k).mean()
        d = k_smooth.rolling(window=smooth_d).mean()
        
        return k_smooth, d
    
    @staticmethod
    def vwap(high, low, close, volume):
        """
        成交量加权平均价格 (Volume Weighted Average Price)
        
        机构投资者常用此作为基准
        """
        typical_price = (high + low + close) / 3
        vwap = (typical_price * volume).cumsum() / volume.cumsum()
        
        return vwap


# 技术分析信号生成器
class TradingSignals:
    """交易信号生成"""
    
    def __init__(self, df):
        """
        参数:
            df: 包含 High, Low, Close, Volume 的 DataFrame
        """
        self.df = df.copy()
        self.ti = TechnicalIndicators()
    
    def rsi_signals(self, oversold=30, overbought=70):
        """RSI 交易信号"""
        rsi = self.ti.rsi(self.df['Close'])
        
        signals = pd.Series(index=self.df.index, dtype=bool)
        signals[rsi < oversold] = 'BUY'  # 超卖
        signals[rsi > overbought] = 'SELL'  # 超买
        signals[(rsi >= oversold) & (rsi <= overbought)] = 'HOLD'
        
        return signals, rsi
    
    def macd_signals(self):
        """MACD 交易信号"""
        macd, signal, hist = self.ti.macd(self.df['Close'])
        
        signals = pd.Series(index=self.df.index, dtype=str)
        
        # MACD 金叉(快线从下往上穿过慢线)
        bullish_cross = (macd > signal) & (macd.shift(1) <= signal.shift(1))
        # MACD 死叉(快线从上往下穿过慢线)
        bearish_cross = (macd < signal) & (macd.shift(1) >= signal.shift(1))
        
        signals[bullish_cross] = 'BUY'
        signals[bearish_cross] = 'SELL'
        signals[~bullish_cross & ~bearish_cross] = 'HOLD'
        
        return signals, macd, signal, hist
    
    def bollinger_signals(self):
        """布林带交易信号"""
        upper, middle, lower = self.ti.bollinger_bands(self.df['Close'])
        
        signals = pd.Series(index=self.df.index, dtype=str)
        
        # 价格触及下轨:超卖,可能买入
        signals[self.df['Close'] < lower] = 'BUY'
        # 价格触及上轨:超买,可能卖出
        signals[self.df['Close'] > upper] = 'SELL'
        signals[(self.df['Close'] >= lower) & (self.df['Close'] <= upper)] = 'HOLD'
        
        return signals, upper, middle, lower
    
    def combined_signals(self):
        """综合多个指标生成信号"""
        rsi_signals, rsi = self.rsi_signals()
        macd_signals, _, _, _ = self.macd_signals()
        bb_signals, _, _, _ = self.bollinger_signals()
        
        # 投票机制:多数指标同意时触发
        votes = pd.DataFrame({
            'RSI': rsi_signals,
            'MACD': macd_signals,
            'BB': bb_signals
        })
        
        # 统计各信号的数量
        buy_count = (votes == 'BUY').sum(axis=1)
        sell_count = (votes == 'SELL').sum(axis=1)
        
        combined = pd.Series(index=self.df.index, dtype=str)
        combined[buy_count >= 2] = 'BUY'
        combined[sell_count >= 2] = 'SELL'
        combined[(buy_count < 2) & (sell_count < 2)] = 'HOLD'
        
        return combined


# 实际应用示例
def generate_analysis_example():
    """生成技术分析示例"""
    # 模拟数据(实际使用中替换为真实数据)
    np.random.seed(42)
    n = 200
    
    dates = pd.date_range(start='2024-01-01', periods=n, freq='D')
    
    # 生成模拟价格序列(带趋势的随机游走)
    returns = np.random.normal(0.001, 0.03, n)
    close = 2000 * np.exp(np.cumsum(returns))
    
    # 生成 OHLCV 数据
    df = pd.DataFrame({
        'Date': dates,
        'High': close * (1 + np.abs(np.random.normal(0, 0.02, n))),
        'Low': close * (1 - np.abs(np.random.normal(0, 0.02, n))),
        'Open': close * (1 + np.random.normal(0, 0.01, n)),
        'Close': close,
        'Volume': np.random.randint(10000, 50000, n)
    })
    df.set_index('Date', inplace=True)
    
    # 计算指标
    ti = TechnicalIndicators()
    
    # RSI
    rsi = ti.rsi(df['Close'])
    
    # MACD
    macd, signal, hist = ti.macd(df['Close'])
    
    # 布林带
    upper, middle, lower = ti.bollinger_bands(df['Close'])
    
    # 生成信号
    ts = TradingSignals(df)
    signals, _, _, _ = ts.macd_signals()
    
    # 打印近期信号
    recent_signals = signals.tail(10)
    print("\nMACD 交易信号 (最近10天):")
    print("-" * 40)
    for date, signal in recent_signals.items():
        if signal != 'HOLD':
            print(f"{date.strftime('%Y-%m-%d')}: {signal}")
    
    # 打印技术指标
    print("\n当前技术指标:")
    print("-" * 40)
    print(f"RSI (14): {rsi.iloc[-1]:.2f}")
    print(f"MACD: {macd.iloc[-1]:.2f}")
    print(f"Signal: {signal.iloc[-1]:.2f}")
    print(f"BB Upper: {upper.iloc[-1]:.2f}")
    print(f"BB Middle: {middle.iloc[-1]:.2f}")
    print(f"BB Lower: {lower.iloc[-1]:.2f}")

3.2 移动平均线交叉策略回测

class MovingAverageCrossover:
    """移动平均线交叉策略"""
    
    def __init__(self, short_period=20, long_period=50):
        self.short_period = short_period
        self.long_period = long_period
    
    def backtest(self, prices, initial_capital=10000, transaction_cost=0.001):
        """
        回测策略
        
        策略逻辑:
        - 短均线上穿长均线 -> 买入
        - 短均线下穿长均线 -> 卖出
        """
        # 计算移动平均
        short_ma = prices.rolling(window=self.short_period).mean()
        long_ma = prices.rolling(window=self.long_period).mean()
        
        # 生成信号
        signal = pd.Series(0, index=prices.index)
        signal[short_ma > long_ma] = 1  # 多头
        signal[short_ma <= long_ma] = 0  # 空头
        
        # 计算持仓变化
        position_change = signal.diff()
        
        # 计算交易
        buys = position_change[position_change > 0]
        sells = position_change[position_change < 0]
        
        # 计算收益
        returns = prices.pct_change()
        strategy_returns = signal.shift(1) * returns
        
        # 扣除交易成本
        num_trades = len(buys) + len(sells)
        total_cost = num_trades * transaction_cost
        strategy_returns = strategy_returns - (transaction_cost * abs(position_change))
        
        # 计算累积收益
        cumulative_returns = (1 + strategy_returns).cumprod()
        cumulative_returns *= initial_capital
        
        # 绩效指标
        total_return = (cumulative_returns.iloc[-1] - initial_capital) / initial_capital
        annual_return = (1 + total_return) ** (252 / len(prices)) - 1
        annual_volatility = strategy_returns.std() * np.sqrt(252)
        sharpe_ratio = (annual_return - 0.03) / annual_volatility
        
        # 最大回撤
        rolling_max = cumulative_returns.cummax()
        drawdown = (cumulative_returns - rolling_max) / rolling_max
        max_drawdown = drawdown.min()
        
        results = {
            'total_return': total_return,
            'annual_return': annual_return,
            'annual_volatility': annual_volatility,
            'sharpe_ratio': sharpe_ratio,
            'max_drawdown': max_drawdown,
            'num_trades': num_trades,
            'total_cost': total_cost,
            'cumulative_returns': cumulative_returns
        }
        
        return results
    
    def optimize_parameters(self, prices, short_range=range(5, 30, 5), 
                           long_range=range(20, 100, 10)):
        """
        参数优化:寻找最优的均线周期组合
        """
        results = []
        
        for short in short_range:
            for long in long_range:
                if short >= long:
                    continue
                
                self.short_period = short
                self.long_period = long
                
                try:
                    result = self.backtest(prices)
                    results.append({
                        'short_period': short,
                        'long_period': long,
                        'sharpe_ratio': result['sharpe_ratio'],
                        'total_return': result['total_return'],
                        'max_drawdown': result['max_drawdown']
                    })
                except:
                    continue
        
        results_df = pd.DataFrame(results)
        best_params = results_df.loc[results_df['sharpe_ratio'].idxmax()]
        
        return results_df, best_params


# 回测示例
def backtest_example():
    """策略回测示例"""
    # 模拟 ETH 价格数据
    np.random.seed(2024)
    n = 500
    
    # 生成带趋势的价格
    trend = np.linspace(0, 0.5, n)
    noise = np.random.normal(0, 0.1, n)
    prices = pd.Series(2000 * np.exp(trend + noise))
    prices.index = pd.date_range(start='2023-01-01', periods=n, freq='D')
    
    # 回测 MA 交叉策略
    strategy = MovingAverageCrossover(short_period=20, long_period=50)
    results = strategy.backtest(prices)
    
    print("=" * 50)
    print("MA 交叉策略回测结果")
    print("=" * 50)
    print(f"总收益率: {results['total_return']:.2%}")
    print(f"年化收益率: {results['annual_return']:.2%}")
    print(f"年化波动率: {results['annual_volatility']:.2%}")
    print(f"夏普比率: {results['sharpe_ratio']:.2f}")
    print(f"最大回撤: {results['max_drawdown']:.2%}")
    print(f"交易次数: {results['num_trades']}")
    print(f"总交易成本: ${results['total_cost']:.2f}")
    
    # 参数优化
    print("\n参数优化中...")
    results_df, best = strategy.optimize_parameters(prices)
    
    print(f"\n最优参数:")
    print(f"  短期均线: {int(best['short_period'])} 天")
    print(f"  长期均线: {int(best['long_period'])} 天")
    print(f"  最优夏普比率: {best['sharpe_ratio']:.2f}")

四、市场数据实务分析

4.1 链上数据分析

class OnChainAnalytics:
    """链上数据分析"""
    
    def __init__(self, provider_url="https://eth-mainnet.g.alchemy.com/v2/demo"):
        """
        初始化
        
        参数:
            provider_url: 以太坊节点 RPC URL
        """
        self.provider_url = provider_url
    
    def get_gas_price_oracle(self):
        """
        获取 Gas 价格预言机数据
        
        返回: (基础费用, 优先费用, 区块容量)
        """
        # 实际实现需要连接以太坊节点
        # 这里使用模拟数据展示结构
        gas_data = {
            'base_fee_per_gas': 20,  # Gwei
            'priority_fee_per_gas': 2,  # Gwei
            'block_gas_limit': 15000000,
            'block_gas_used': 12000000,
            'utilization': 0.80
        }
        return gas_data
    
    def get_eth_supply(self):
        """
        获取 ETH 供应量数据
        
        返回: 供应量统计
        """
        supply_data = {
            'total_supply': 120700000,  # ETH
            'circulating_supply': 73500000,
            'staked_supply': 45000000,
            'staking_ratio': 0.373,
            'burned_supply': 3200000
        }
        return supply_data
    
    def get_defi_tvl(self):
        """
        获取 DeFi TVL 数据
        
        返回: 各协议 TVL
        """
        tvl_data = {
            'Aave': 12500000000,  # $12.5B
            'MakerDAO': 8500000000,
            'Uniswap': 7200000000,
            'Curve': 5800000000,
            'Lido': 15000000000,
            'Total': 52000000000
        }
        return tvl_data
    
    def calculate_network_value(self, price, supply):
        """
        计算网络价值指标
        
        参数:
            price: ETH 价格
            supply: 流通供应量
        """
        network_value = price * supply
        
        # MVRV 比率 (Market Value to Realized Value)
        # 需获取已实现市值,这里使用简化计算
        realized_value = network_value * 0.85  # 简化估算
        mvrv_ratio = network_value / realized_value
        
        return {
            'network_value': network_value,
            'realized_value': realized_value,
            'mvrv_ratio': mvrv_ratio,
            'interpretation': 'overvalued' if mvrv_ratio > 3 else 'undervalued' if mvrv_ratio < 1 else 'fair'
        }
    
    def get_whale_activity(self, large_transaction_threshold=1000000):
        """
        分析巨鲸活动
        
        参数:
            large_transaction_threshold: 大额交易阈值(美元)
            
        返回: 巨鲸活动指标
        """
        # 实际实现需要获取链上交易数据
        activity = {
            'large_transactions_24h': 145,
            'total_volume_24h': 8500000000,
            'exchange_inflow_24h': 125000,
            'exchange_outflow_24h': 98000,
            'net_flow': -27000,  # 净流出
            'whale_accumulation_score': 0.65  # 0-1, 高值表示积累
        }
        return activity


# 链上分析报告生成
def generate_onchain_report():
    """生成链上分析报告"""
    analyzer = OnChainAnalytics()
    
    # 获取各类数据
    gas = analyzer.get_gas_price_oracle()
    supply = analyzer.get_eth_supply()
    tvl = analyzer.get_defi_tvl()
    whale = analyzer.get_whale_activity()
    
    # 计算网络价值
    price = 2500  # 假设 ETH 价格
    network = analyzer.calculate_network_value(price, supply['circulating_supply'])
    
    print("=" * 60)
    print("以太坊链上分析报告")
    print("=" * 60)
    
    print("\n【Gas 费用状况】")
    print(f"  基础费用: {gas['base_fee_per_gas']} Gwei")
    print(f"  优先费用: {gas['priority_fee_per_gas']} Gwei")
    print(f"  区块利用率: {gas['utilization']:.1%}")
    
    print("\n【ETH 供应状况】")
    print(f"  总供应量: {supply['total_supply']/1e6:.2f}M ETH")
    print(f"  流通供应: {supply['circulating_supply']/1e6:.2f}M ETH")
    print(f"  质押数量: {supply['staked_supply']/1e6:.2f}M ETH")
    print(f"  质押率: {supply['staking_ratio']:.1%}")
    
    print("\n【DeFi TVL 分布】")
    for protocol, value in tvl.items():
        if protocol != 'Total':
            print(f"  {protocol}: ${value/1e9:.2f}B")
    print(f"  总计: ${tvl['Total']/1e9:.2f}B")
    
    print("\n【巨鲸活动】")
    print(f"  24h 大额交易: {whale['large_transactions_24h']} 笔")
    print(f"  24h 净流入交易所: {whale['net_flow']:,.0f} ETH")
    print(f"  积累评分: {whale['whale_accumulation_score']:.2f}")
    
    print("\n【估值指标】")
    print(f"  网络价值: ${network['network_value']/1e9:.2f}B")
    print(f"  MVRV 比率: {network['mvrv_ratio']:.2f}")
    print(f"  估值状态: {network['interpretation']}")

4.2 市场情绪量化分析

class SentimentAnalyzer:
    """市场情绪分析器"""
    
    def __init__(self):
        self.fear_greed_index = None
    
    def calculate_fear_greed_index(self, volatility, volume, social, dominance):
        """
        计算恐懼與貪婪指數
        
        参数:
            volatility: 波动率指标
            volume: 交易量指标
            social: 社交媒体情绪
            market_dominance: 市场主导地位
        
        返回: 0-100 的指数值
        """
        # 归一化各指标
        vol_score = self._normalize_volatility(volatility)
        vol_score = 100 - vol_score  # 低波动 = 贪婪
        
        vol_score = np.clip(vol_score, 0, 100)
        
        # 交易量评分(高交易量通常表示贪婪)
        vol_ratio = volume / volume.rolling(30).mean()
        vol_score = 50 + (vol_ratio - 1) * 25
        vol_score = np.clip(vol_score, 0, 100)
        
        # 社交媒体评分(需要实际 API 数据)
        social_score = 50  # 简化
        social_score = np.clip(social_score, 0, 100)
        
        # 主导地位评分(BTC 主导下降 = 贪婪)
        dom_score = 50 + (50 - dominance) * 0.5
        dom_score = np.clip(dom_score, 0, 100)
        
        # 综合评分
        weights = [0.25, 0.25, 0.25, 0.25]
        index = (vol_score * weights[0] + vol_score * weights[1] + 
                social_score * weights[2] + dom_score * weights[3])
        
        return index
    
    def _normalize_volatility(self, volatility):
        """波动率归一化"""
        vol_30d_avg = volatility.rolling(30).mean()
        vol_90d_avg = volatility.rolling(90).mean()
        
        normalized = (volatility - vol_90d_avg) / vol_90d_avg
        normalized = 50 + normalized * 100
        
        return np.clip(normalized, 0, 100)
    
    def get_regime_classification(self, returns, volatility):
        """
        市场状态分类
        
        返回: (市场状态, 建议策略)
        """
        recent_return = returns.tail(20).mean()
        recent_vol = volatility.tail(20).mean()
        historical_vol = volatility.rolling(90).mean()
        
        if recent_vol > historical_vol * 1.5:
            regime = "HIGH_VOLATILITY"
            strategy = "降低仓位,使用期权保护"
        elif recent_return > 0.05:
            regime = "BULL"
            strategy = "顺势而为适度加仓"
        elif recent_return < -0.05:
            regime = "BEAR"
            strategy = "现金为王,等待机会"
        else:
            regime = "CONSOLIDATION"
            strategy = "区间操作,高抛低吸"
        
        return regime, strategy
    
    def correlation_analysis(self, assets_returns):
        """
        资产相关性分析
        
        返回: 相关性矩阵和聚类结果
        """
        corr_matrix = assets_returns.corr()
        
        # 简单聚类:相关性 > 0.7 的资产归为一类
        clusters = {}
        for i, asset in enumerate(corr_matrix.columns):
            if asset not in clusters:
                clusters[asset] = f"Cluster_{i}"
            
            for j, other in enumerate(corr_matrix.columns):
                if i != j and corr_matrix.loc[asset, other] > 0.7:
                    clusters[other] = clusters[asset]
        
        return corr_matrix, clusters


# 情绪分析报告
def sentiment_report_example():
    """生成情绪分析报告示例"""
    
    # 模拟数据
    np.random.seed(42)
    dates = pd.date_range(start='2024-01-01', periods=180, freq='D')
    
    prices = pd.Series(2000 * np.exp(np.cumsum(np.random.normal(0.001, 0.03, 180))), 
                       index=dates)
    returns = prices.pct_change().dropna()
    volatility = returns.rolling(14).std() * np.sqrt(365)
    
    analyzer = SentimentAnalyzer()
    
    # 市场状态分类
    regime, strategy = analyzer.get_regime_classification(returns, volatility)
    
    print("=" * 50)
    print("市场情绪分析报告")
    print("=" * 50)
    print(f"\n当前市场状态: {regime}")
    print(f"建议策略: {strategy}")
    
    # 资产相关性分析
    assets = pd.DataFrame({
        'ETH': returns,
        'BTC': returns * 0.9 + np.random.normal(0, 0.01, len(returns)),
        'SOL': returns * 1.2 + np.random.normal(0, 0.02, len(returns)),
        'USDC': np.random.normal(0, 0.0001, len(returns))
    })
    
    corr, clusters = analyzer.correlation_analysis(assets)
    
    print("\n资产相关性矩阵:")
    print(corr.round(3))

五、实战投资决策系统

5.1 综合评分模型

class InvestmentDecisionSystem:
    """综合投资决策系统"""
    
    def __init__(self):
        self.weights = {
            'technical': 0.25,
            'fundamental': 0.30,
            'sentiment': 0.20,
            'onchain': 0.25
        }
    
    def technical_score(self, prices):
        """
        技术面评分 (0-100)
        
        考虑: MA趋势、RSI、MACD、成交量
        """
        # 计算各指标
        sma_20 = prices.rolling(20).mean()
        sma_50 = prices.rolling(50).mean()
        sma_200 = prices.rolling(200).mean() if len(prices) > 200 else sma_50
        
        # 趋势评分
        trend_score = 0
        if prices.iloc[-1] > sma_20.iloc[-1]:
            trend_score += 25
        if sma_20.iloc[-1] > sma_50.iloc[-1]:
            trend_score += 25
        if sma_50.iloc[-1] > sma_200.iloc[-1]:
            trend_score += 20
        
        # RSI 评分
        delta = prices.diff()
        gain = delta.where(delta > 0, 0).rolling(14).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
        rs = gain / loss
        rsi = 100 - (100 / (1 + rs))
        
        rsi_score = 0
        if 30 < rsi.iloc[-1] < 70:
            rsi_score = 15  # 中性最优
        elif rsi.iloc[-1] <= 30:
            rsi_score = 10  # 超卖,可能反弹
        else:
            rsi_score = 5  # 超买
        
        # 综合技术得分
        total_score = trend_score + rsi_score
        return min(100, total_score)
    
    def fundamental_score(self):
        """
        基本面评分 (0-100)
        
        考虑: 质押率、燃烧率、TVL、机构采用
        """
        scores = []
        
        # 质押率(高质押率 = 高分)
        staking_ratio = 0.37
        staking_score = min(100, staking_ratio * 200)
        scores.append(staking_score * 0.2)
        
        # 燃烧率(净通缩 = 高分)
        burn_rate = 0.02  # 假设年化2%
        burn_score = min(100, abs(burn_rate) * 2000)
        scores.append(burn_score * 0.2)
        
        # DeFi TVL 增长
        tvl_growth = 0.15  # 假设年化15%
        tvl_score = min(100, (1 + tvl_growth) * 30)
        scores.append(tvl_score * 0.3)
        
        # 机构采用
        institutional_score = 70  # 假设
        scores.append(institutional_score * 0.3)
        
        return sum(scores)
    
    def sentiment_score(self):
        """
        情绪面评分 (0-100)
        
        使用恐懼與貪婪指数
        """
        fear_greed = 55  # 假设当前值
        
        # 极端恐惧可能是买入机会
        if fear_greed < 25:
            score = 80  # 买入机会
        elif fear_greed < 45:
            score = 60  # 谨慎买入
        elif fear_greed < 65:
            score = 50  # 中性
        elif fear_greed < 85:
            score = 40  # 谨慎
        else:
            score = 20  # 极端贪婪,风险积累
        
        return score
    
    def onchain_score(self):
        """
        链上面评分 (0-100)
        
        考虑: MVRV、交易所净流量、巨鲸活动
        """
        scores = []
        
        # MVRV (2-3 为中性)
        mvrv = 2.2
        if 1.5 < mvrv < 3:
            mvrv_score = 80
        elif 1 < mvrv < 4:
            mvrv_score = 60
        else:
            mvrv_score = 40
        scores.append(mvrv_score * 0.35)
        
        # 交易所净流量(净流出 = 积累 = 高分)
        net_flow = -25000  # ETH 净流出
        if net_flow < -10000:
            flow_score = 80
        elif net_flow < 0:
            flow_score = 60
        else:
            flow_score = 40
        scores.append(flow_score * 0.35)
        
        # 活跃地址增长
        addr_growth = 0.08
        addr_score = min(100, addr_growth * 500 + 50)
        scores.append(addr_score * 0.3)
        
        return sum(scores)
    
    def overall_score(self, prices):
        """
        综合评分
        
        返回: (总分, 评级, 建议)
        """
        tech = self.technical_score(prices)
        fund = self.fundamental_score()
        sent = self.sentiment_score()
        onchain = self.onchain_score()
        
        total = (tech * self.weights['technical'] +
                fund * self.weights['fundamental'] +
                sent * self.weights['sentiment'] +
                onchain * self.weights['onchain'])
        
        # 评级
        if total >= 70:
            rating = "强烈推荐"
            action = "增加仓位"
        elif total >= 55:
            rating = "推荐"
            action = "维持或小幅增持"
        elif total >= 40:
            rating = "中性"
            action = "保持当前仓位"
        elif total >= 25:
            rating = "谨慎"
            action = "考虑减仓"
        else:
            rating = "不推荐"
            action = "卖出或观望"
        
        return {
            'total_score': round(total, 1),
            'rating': rating,
            'action': action,
            'breakdown': {
                'technical': round(tech, 1),
                'fundamental': round(fund, 1),
                'sentiment': round(sent, 1),
                'onchain': round(onchain, 1)
            }
        }


# 综合决策示例
def investment_decision_example():
    """投资决策示例"""
    
    # 模拟价格数据
    np.random.seed(42)
    dates = pd.date_range(start='2024-01-01', periods=250, freq='D')
    prices = pd.Series(2000 * np.exp(np.cumsum(np.random.normal(0.001, 0.025, 250))),
                       index=dates)
    
    # 创建决策系统
    decision_system = InvestmentDecisionSystem()
    
    # 获取综合评分
    result = decision_system.overall_score(prices)
    
    print("=" * 60)
    print("以太坊投资决策报告")
    print("=" * 60)
    
    print("\n【评分明细】")
    for category, score in result['breakdown'].items():
        bar = '█' * int(score / 5) + '░' * (20 - int(score / 5))
        category_name = {
            'technical': '技术面',
            'fundamental': '基本面',
            'sentiment': '情绪面',
            'onchain': '链上面'
        }
        print(f"  {category_name.get(category, category):8s}: {bar} {score:.1f}")
    
    print("\n【综合评分】")
    print(f"  总分: {result['total_score']} / 100")
    print(f"  评级: {result['rating']}")
    print(f"  建议: {result['action']}")

5.2 投资组合监控仪表板

class PortfolioDashboard:
    """投资组合监控仪表板"""
    
    def __init__(self, holdings, prices):
        """
        参数:
            holdings: 持仓字典 {'资产': 数量}
            prices: 价格字典 {'资产': 价格}
        """
        self.holdings = holdings
        self.prices = prices
    
    def portfolio_value(self):
        """计算组合总价值"""
        total = sum(amount * self.prices.get(asset, 0) 
                   for asset, amount in self.holdings.items())
        return total
    
    def allocation(self):
        """资产配置"""
        total = self.portfolio_value()
        return {asset: (amount * self.prices.get(asset, 0) / total) 
                for asset, amount in self.holdings.items()}
    
    def daily_pnl(self, yesterday_prices):
        """当日盈亏"""
        current_value = self.portfolio_value()
        
        yesterday_value = sum(
            amount * yesterday_prices.get(asset, 0)
            for asset, amount in self.holdings.items()
        )
        
        pnl = current_value - yesterday_value
        pnl_pct = pnl / yesterday_value if yesterday_value > 0 else 0
        
        return pnl, pnl_pct
    
    def risk_metrics(self, returns_df):
        """风险指标"""
        portfolio_returns = pd.Series()
        total_value = self.portfolio_value()
        
        for asset, weight in self.allocation().items():
            if asset in returns_df.columns:
                portfolio_returns += returns_df[asset] * weight
        
        var_95 = portfolio_returns.quantile(0.05)
        max_dd = ((1 + portfolio_returns).cumprod() - 
                 (1 + portfolio_returns).cumprod().cummax()).min()
        
        return {
            'VaR 95%': var_95,
            'Max Drawdown': max_dd,
            'Volatility (annual)': portfolio_returns.std() * np.sqrt(365)
        }
    
    def generate_report(self, yesterday_prices=None, returns_df=None):
        """生成完整报告"""
        print("=" * 70)
        print("投资组合监控仪表板")
        print("=" * 70)
        
        # 基本信息
        print("\n【持仓概况】")
        total = self.portfolio_value()
        for asset, amount in self.holdings.items():
            value = amount * self.prices.get(asset, 0)
            weight = value / total
            print(f"  {asset:8s}: {amount:>10.4f} @ ${self.prices.get(asset, 0):>8.2f} "
                  f"= ${value:>12,.2f} ({weight:>6.1%})")
        
        print(f"\n  {'组合总价值':12s}: ${total:>15,.2f}")
        
        # 盈亏
        if yesterday_prices:
            pnl, pnl_pct = self.daily_pnl(yesterday_prices)
            print(f"\n【当日盈亏】")
            print(f"  PnL: ${pnl:>12,.2f} ({pnl_pct:>+.2%})")
        
        # 风险指标
        if returns_df is not None:
            risk = self.risk_metrics(returns_df)
            print(f"\n【风险指标】")
            for metric, value in risk.items():
                if 'VaR' in metric:
                    print(f"  {metric:20s}: {value:>+.2%}")
                else:
                    print(f"  {metric:20s}: {value:>+.2%}")


# 仪表板示例
def dashboard_example():
    """仪表板使用示例"""
    
    # 持仓
    holdings = {
        'ETH': 10.0,
        'BTC': 0.5,
        'USDC': 20000,
        'UNI': 500,
        'ARB': 2000
    }
    
    # 当前价格
    prices = {
        'ETH': 2520.0,
        'BTC': 68500.0,
        'USDC': 1.0,
        'UNI': 7.50,
        'ARB': 0.85
    }
    
    # 昨日价格
    yesterday_prices = {
        'ETH': 2480.0,
        'BTC': 67200.0,
        'USDC': 1.0,
        'UNI': 7.30,
        'ARB': 0.82
    }
    
    # 生成报告
    dashboard = PortfolioDashboard(holdings, prices)
    dashboard.generate_report(yesterday_prices)

结论与实践建议

本文提供了以太坊投资策略的完整量化框架,涵盖投资组合优化、风险管理、技术分析和市场监控四大核心领域。投资者可以根据自身需求,选择性地应用这些工具和方法。

关键实践要点包括:

第一,系统化投资决策。通过综合评分模型,将技术面、基本面、情绪面和链上数据整合为统一的决策框架,避免情绪化操作。

第二,重视风险控制。无论使用何种策略,都应设置明确的止损点位和仓位上限。VaR 和压力测试可以帮助投资者了解潜在损失。

第三,持续监控与调整。市场环境不断变化,投资组合需要定期审视和调整。再平衡策略可以帮助维持目标风险水平。

第四,数据驱动决策。尽可能使用量化工具和数据分析支持投资决策,减少主观判断的偏差。

需要强调的是,本文提供的代码和模型仅供参考,实际应用时需要根据个人情况和市场变化进行调整。加密货币投资风险极高,投资者应仅使用可承受损失的资金,并进行充分的独立研究。


参考资源

  1. 以太坊官方文档 - ethereum.org
  2. Dune Analytics - 数据分析平台
  3. Glassnode Studio - 链上分析工具
  4. CryptoQuant - 链上数据分析
  5. Yahoo Finance - 市场数据
  6. Pandas 和 NumPy 文档 - Python 数据分析

延伸閱讀與來源

這篇文章對您有幫助嗎?

評論

發表評論

注意:由於這是靜態網站,您的評論將儲存在本地瀏覽器中,不會公開顯示。

目前尚無評論,成為第一個發表評論的人吧!