顶级期货分析师知识体系

# 持有成本模型 - 期货定价 # F = S * exp((r + u - y) * T) import numpy as np import pandas as pd def cost_of_carry_pricing(S, r, u, y, T): """ 持有成本模型计算期货理论价格 参数: S=现货价格, r=无风险利率, u=仓储费率, y=便利收益率, T=到期时间(年) """ F = S * np.exp((r + u - y) * T) components = pd.DataFrame({ '参数': ['现货价格', '无风险利率', '仓储费率', '便利收益率', '到期时间', '期货价格'], '值': [S, f"{r*100}%", f"{u*100}%", f"{y*100}%", f"{T*12}个月", round(F, 4)] }) print("═══ 持有成本模型定价 ═══") print(components.to_string(index=False)) print(f"\n基差 Basis = S - F = {round(S - F, 4)}") return F # 示例：沪铜期货定价 F = cost_of_carry_pricing( S=72000, # 现货价格（元/吨） r=0.025, # 无风险利率 2.5% u=0.008, # 仓储费率 0.8% y=0.005, # 便利收益率 0.5% T=0.5 # 6个月到期 ) # 基差分析 def basis_analysis(S, F): if F > S: print("正向市场 (Contango): 期货 > 现货") else: print("反向市场 (Backwardation): 现货 > 期货") basis_analysis(72000, F)

# 移动平均线系统 - Python实现 import numpy as np import pandas as pd # 生成示例价格数据 np.random.seed(42) price = np.cumsum(np.random.normal(0, 50, 250)) + 5000 dates = pd.date_range('2025-01-01', periods=250) df = pd.DataFrame({'price': price}, index=dates) # 计算各类均线 df['SMA_5'] = df['price'].rolling(5).mean() df['SMA_20'] = df['price'].rolling(20).mean() df['SMA_60'] = df['price'].rolling(60).mean() df['EMA_12'] = df['price'].ewm(span=12, adjust=False).mean() df['EMA_26'] = df['price'].ewm(span=26, adjust=False).mean() # MACD df['MACD'] = df['EMA_12'] - df['EMA_26'] df['Signal'] = df['MACD'].ewm(span=9, adjust=False).mean() df['Histogram'] = df['MACD'] - df['Signal'] # 布林带 df['BB_MID'] = df['price'].rolling(20).mean() df['BB_STD'] = df['price'].rolling(20).std() df['BB_UP'] = df['BB_MID'] + 2 * df['BB_STD'] df['BB_LO'] = df['BB_MID'] - 2 * df['BB_STD'] # 金叉/死叉信号 df['Signal_Cross'] = 0 df.loc[df['SMA_5'] > df['SMA_20'], 'Signal_Cross'] = 1 df['Golden_Cross'] = df['Signal_Cross'].diff() == 1 df['Death_Cross'] = df['Signal_Cross'].diff() == -1 print(df[dropna()].tail(10))

# RSI 相对强弱指数 - Python实现 import pandas as pd import numpy as np def calculate_rsi(series, period=14): """计算RSI指标""" delta = series.diff() gain = delta.where(delta > 0, 0) loss = (-delta).where(delta < 0, 0) avg_gain = gain.ewm(alpha=1/period, min_periods=period).mean() avg_loss = loss.ewm(alpha=1/period, min_periods=period).mean() rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) return rsi # 使用示例 df['RSI_14'] = calculate_rsi(df['price']) # 信号判断 df['RSI_Signal'] = None df.loc[df['RSI_14'] > 70, 'RSI_Signal'] = '超买' df.loc[df['RSI_14'] < 30, 'RSI_Signal'] = '超卖' df.loc[(df['RSI_14'] >= 30) & (df['RSI_14'] <= 70), 'RSI_Signal'] = '中性' print(df[['price', 'RSI_14', 'RSI_Signal']].tail(10))

# 供需平衡表分析 - Python实现 import pandas as pd import numpy as np # 构建示例供需平衡表（以沪铜为例） data = { '年份': list(range(2019, 2026)), '期初库存': [180,165,155,140,130,120,135], '产量': [985,1002,1049,1106,1150,1198,1240], '进口量': [380,450,550,530,480,510,525], '消费量': [1180,1380,1480,1510,1520,1560,1600], '出口量': [5,8,6,10,8,7,9] } df = pd.DataFrame(data) df['总供给'] = df['期初库存'] + df['产量'] + df['进口量'] df['总需求'] = df['消费量'] + df['出口量'] df['期末库存'] = df['总供给'] - df['总需求'] df['供需缺口'] = df['总供给'] - df['总需求'] df['库存消费比'] = (df['期末库存'] / df['消费量'] * 100).round(1) print(df.to_string(index=False)) # 导出Excel df.to_excel('供需平衡表_沪铜.xlsx', index=False)

# ARIMA模型 - Python实现 import numpy as np import pandas as pd from statsmodels.tsa.arima.model import ARIMA from statsmodels.tsa.stattools import adfuller from statsmodels.graphics.tsaplots import plot_acf, plot_pacf import matplotlib.pyplot as plt # 生成示例数据 np.random.seed(42) price = np.cumsum(np.random.normal(5, 30, 500)) + 5000 # ADF平稳性检验 adf_result = adfuller(price) print(f"ADF统计量: {adf_result[0]:.4f}") print(f"p值: {adf_result[1]:.4f}") # 差分使序列平稳 price_diff = np.diff(price) # 拟合ARIMA模型 model = ARIMA(price, order=(2, 1, 2)) fitted = model.fit() print(fitted.summary()) # 预测 forecast = fitted.get_forecast(steps=20) fc_mean = forecast.predicted_mean fc_ci = forecast.conf_int(alpha=0.05) # 绘制预测结果 fig, ax = plt.subplots(figsize=(12, 5)) ax.plot(price[-100:], label='实际价格') ax.plot(range(400, 420), fc_mean, 'r--', label='预测') ax.fill_between(range(400, 420), fc_ci.iloc[:, 0], fc_ci.iloc[:, 1], alpha=0.2, color='red') ax.legend() ax.set_title('ARIMA期货价格预测') plt.show()

# GARCH(1,1)模型 - Python实现 import numpy as np from arch import arch_model import matplotlib.pyplot as plt # 计算日收益率 returns = np.diff(np.log(price)) * 100 # 拟合GARCH(1,1) model = arch_model(returns, vol='Garch', p=1, q=1, mean='Constant', dist='normal') fitted = model.fit(update_freq=5) print(fitted.summary()) # 提取参数 omega = fitted.params['omega'] alpha = fitted.params['alpha[1]'] beta = fitted.params['beta[1]'] print(f"\nω = {omega:.6f}") print(f"α = {alpha:.4f}") print(f"β = {beta:.4f}") print(f"α+β = {alpha+beta:.4f}") # 波动率预测 forecast = fitted.forecast(horizon=20) sigma_fc = np.sqrt(forecast.variance[-1:].values[0]) print(f"未来20日波动率预测: {sigma_fc.round(4)}")

# 协整检验与配对套利 - Python实现 import numpy as np import pandas as pd from statsmodels.tsa.stattools import adfuller, coint from statsmodels.regression.linear_model import OLS from statsmodels.tools import add_constant import matplotlib.pyplot as plt # 模拟两个协整序列 np.random.seed(42) n = 500 common = np.cumsum(np.random.normal(0.1, 1, n)) y1 = 4000 + common * 10 + np.random.normal(0, 20, n) y2 = 3800 + common * 9 + np.random.normal(0, 18, n) # Step 1: OLS回归获取对冲比 X = add_constant(y2) model = OLS(y1, X).fit() hedge_ratio = model.params[1] print(f"对冲比 (Hedge Ratio): {hedge_ratio:.4f}") # Step 2: 残差ADF检验 residuals = model.resid adf_stat, adf_p, _, _, _, _ = adfuller(residuals) print(f"残差ADF p值: {adf_p:.4f}") # 构建套利信号 spread = y1 - hedge_ratio * y2 z_score = (spread - spread.mean()) / spread.std() df_signal = pd.DataFrame({ 'y1': y1, 'y2': y2, 'spread': spread, 'z_score': z_score }) df_signal['signal'] = None df_signal.loc[z_score > 2, 'signal'] = '做空价差' df_signal.loc[z_score < -2, 'signal'] = '做多价差' # 回测统计 print(f"\n═══ 配对套利统计 ═══") print(f"价差均值: {spread.mean():.2f}") print(f"价差标准差: {spread.std():.2f}") print(f"做空信号: {(df_signal['signal']=='做空价差').sum()}次") print(f"做多信号: {(df_signal['signal']=='做多价差').sum()}次")

# 随机森林期货涨跌预测 - Python实现 import numpy as np import pandas as pd from sklearn.ensemble import RandomForestClassifier from sklearn.model_selection import train_test_split from sklearn.metrics import classification_report, confusion_matrix # 准备数据 np.random.seed(42) n = 500 close = np.cumsum(np.random.normal(0, 50, n)) + 5000 volume = np.abs(np.random.normal(100000, 30000, n)) df = pd.DataFrame({'close': close, 'volume': volume}) df['returns'] = df['close'].pct_change() df['volatility'] = df['returns'].rolling(20).std() df['ma5_gap'] = (df['close'].rolling(5).mean() - df['close']) / df['close'] df['ma20_gap'] = (df['close'].rolling(20).mean() - df['close']) / df['close'] df['rsi'] = calculate_rsi(df['close']) df['vol_chg'] = df['volume'].pct_change() df = df.dropna() # 标签：次日涨跌 df['label'] = (df['returns'].shift(-1) > 0).astype(int) features = ['returns', 'volatility', 'ma5_gap', 'ma20_gap', 'rsi', 'vol_chg'] X = df[features].iloc[:-1] y = df['label'].iloc[:-1] X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, shuffle=False) # 训练 rf = RandomForestClassifier(n_estimators=500, max_features=3, random_state=42) rf.fit(X_train, y_train) # 预测与评估 y_pred = rf.predict(X_test) print(classification_report(y_test, y_pred, target_names=['跌', '涨'])) # 特征重要性 importance = pd.Series(rf.feature_importances_, index=features).sort_values(ascending=False) print("\n特征重要性:") print(importance)

# VaR计算 - Python实现 import numpy as np from scipy.stats import norm # 模拟收益率 np.random.seed(42) returns = np.random.normal(0.0002, 0.015, 1000) # 1. 历史模拟法 var_hist = np.percentile(returns, 5) print(f"历史模拟法 95% VaR: {var_hist*100:.4f}%") # 2. 参数法 mu, sigma = returns.mean(), returns.std() var_param = mu - 1.645 * sigma print(f"参数法 95% VaR: {var_param*100:.4f}%") # 3. CVaR (Expected Shortfall) cvar = returns[returns <= var_hist].mean() print(f"CVaR (Expected Shortfall): {cvar*100:.4f}%") # 4. 蒙特卡洛模拟法 n_sims = 10000 sim_returns = np.random.normal(mu, sigma, n_sims) var_mc = np.percentile(sim_returns, 5) print(f"蒙特卡洛 95% VaR: {var_mc*100:.4f}%")

# 凯利公式仓位管理 - Python实现 def kelly_criterion(win_rate, avg_win, avg_loss): """ 计算凯利最优仓位比例 """ odds = avg_win / avg_loss # 赔率 q = 1 - win_rate kelly = (odds * win_rate - q) / odds half_kelly = kelly / 2 print("═══ 凯利公式仓位计算 ═══") print(f"胜率: {win_rate*100}%") print(f"赔率(盈亏比): {odds:.2f}") print(f"最优仓位(凯利): {kelly*100:.1f}%") print(f"建议仓位(半凯利): {half_kelly*100:.1f}%") if kelly <= 0: print("⚠️ 凯利值为负，不应交易此策略") return kelly, half_kelly # 示例 kelly, half_kelly = kelly_criterion(win_rate=0.55, avg_win=3000, avg_loss=2000)

# Black-Scholes 期权定价 - Python实现 import numpy as np from scipy.stats import norm def bs_pricing(S, K, T, r, sigma, option_type='call'): """ Black-Scholes期权定价与希腊字母 """ d1 = (np.log(S/K) + (r + sigma**2/2)*T) / (sigma*np.sqrt(T)) d2 = d1 - sigma*np.sqrt(T) if option_type == 'call': price = S*norm.cdf(d1) - K*np.exp(-r*T)*norm.cdf(d2) else: price = K*np.exp(-r*T)*norm.cdf(-d2) - S*norm.cdf(-d1) delta = norm.cdf(d1) if option_type=='call' else norm.cdf(d1)-1 gamma = norm.pdf(d1) / (S*sigma*np.sqrt(T)) vega = S*norm.pdf(d1)*np.sqrt(T) / 100 theta = -(S*norm.pdf(d1)*sigma) / (2*np.sqrt(T)) / 365 print(f"═══ Black-Scholes定价 ═══") print(f"期权价格: {price:.2f}") print(f"Delta: {delta:.4f}") print(f"Gamma: {gamma:.6f}") print(f"Vega: {vega:.4f}") print(f"Theta: {theta:.4f}") return {'price': price, 'delta': delta, 'gamma': gamma, 'vega': vega, 'theta': theta} # 示例：沪铜看涨期权 result = bs_pricing(S=72000, K=74000, T=90/365, r=0.025, sigma=0.25, option_type='call')

# 跨期套利回测 - Python实现 import numpy as np import pandas as pd # 模拟近月和远月合约 np.random.seed(42) n = 250 near = np.cumsum(np.random.normal(2, 40, n)) + 3800 far = near + np.random.normal(80, 20, n) spread = far - near z = (spread - spread.mean()) / spread.std() # 交易信号 position = np.where(z > 1.5, -1, np.where(z < -1.5, 1, 0)) spread_ret = np.diff(spread, prepend=spread[0]) strategy_ret = np.roll(position, 1) * spread_ret strategy_ret[0] = 0 # 绩效 total = strategy_ret.sum() sharpe = (strategy_ret.mean() / strategy_ret.std()) * np.sqrt(252) cumret = np.cumsum(strategy_ret) max_dd = np.max(np.maximum.accumulate(cumret) - cumret) win_rate = (strategy_ret > 0).mean() print(f"═══ 策略绩效 ═══") print(f"总收益: {total:.2f}") print(f"夏普比率: {sharpe:.2f}") print(f"最大回撤: {max_dd:.2f}") print(f"胜率: {win_rate*100:.1f}%")

# NLP新闻情绪分析 - Python实现 import re # 情绪词典 POSITIVE = ["上涨", "利好", "突破", "支撑", "反弹", "增长", "需求旺盛", "供应紧张", "超预期"] NEGATIVE = ["下跌", "利空", "跌破", "压力", "回调", "收缩", "需求疲弱", "库存高企", "不及预期"] def analyze_sentiment(text): """分析文本情绪分数""" pos = sum(1 for w in POSITIVE if w in text) neg = sum(1 for w in NEGATIVE if w in text) total = pos + neg if total == 0: return 0, "中性" score = (pos - neg) / total label = "看多" if score > 0.3 else ("看空" if score < -0.3 else "中性") return round(score, 2), label # 示例 news_list = [ "铜价突破关键阻力位，需求旺盛支撑上涨", "库存高企叠加需求疲弱，螺纹钢承压下跌", "原油供应紧张，市场预期OPEC将延长减产" ] for news in news_list: score, label = analyze_sentiment(news) print(f"新闻: {news}") print(f"情绪: {label} (分数: {score})\n")