pip install tushare yfinance pandas numpy talib matplotlibpip install backtrader # 专业的回测框架pip install ccxt # 加密货币交易所接口pip install sqlalchemy # 数据库管理摘要：import talibdefcalculate_bollinger_bands(df, window=20, num_std=2): df['MA']= df['close'].rolling(window).mean df['STD']= df['clos
布林带（Bollinger Bands）：
import talibdefcalculate_bollinger_bands(df, window=20, num_std=2): df['MA']= df['close'].rolling(window).mean df['STD']= df['close'].rolling(window).std df['Upper_Band']= df['MA']+(df['STD']* num_std) df['Lower_Band']= df['MA']-(df['STD']* num_std) return df布林带策略是一种基于波动率的交易策略。当价格触及布林带上轨时，卖出；当价格触及下轨时，买入：
def bollinger_bands_strategy(df): df = calculate_bollinger_bands(df) df['Signal']=0 df.loc[df['close']> df['Upper_Band'],'Signal']=-1 df.loc[df['close']均值回归策略假设价格会回归到其均值水平。当价格偏离均值时，我们可以通过计算Z-score来判断买入或卖出时机：
def mean_reversion_strategy(df, window=20, z_threshold=2): df['MA']= df['close'].rolling(window).mean df['STD']= df['close'].rolling(window).std df['Z_Score']=(df['close']- df['MA'])/ df['STD'] df['Signal']=0 df.loc[df['Z_Score']> z_threshold,'Signal']=-1 df.loc[df['Z_Score']机器学习在量化交易中的应用越来越广泛。我们可以使用简单的线性回归模型来预测价格走势：
from sklearn.linear_model import LinearRegressiondefmachine_learning_strategy(df, train_size=0.8): df['Target']= df['close'].shift(-1) df.dropna(inplace=True) X = df[['open','high','low','close','volume']] y = df['Target'] train_size =int(len(df)* train_size) X_train, X_test = X[:train_size], X[train_size:] y_train, y_test = y[:train_size], y[train_size:] model = LinearRegression model.fit(X_train, y_train) df['Prediction']= model.predict(X) df['Signal']=0 df.loc[df['Prediction']> df['close'],'Signal']=1 df.loc[df['Prediction']动态止损可以根据市场波动调整止损点，避免过早止损：
def dynamic_stop_loss(df, stop_loss_pct=0.05): df['Stop_Loss'] = df['close'] * (1 - stop_loss_pct) df['Stop_Loss'] = df['Stop_Loss'].rolling(5).max return df在多资产组合中，我们需要平衡不同资产的风险暴露：
def portfolio_risk_management(df_list, weights): portfolio_return = sum([df['close'].pct_change.mean * weight for df, weight in zip(df_list, weights)]) portfolio_risk = sum([df['close'].pct_change.std * weight for df, weight in zip(df_list, weights)]) return portfolio_return, portfolio_riskclass MultiStrategyBackTest: def__init__(self, df, strategies, initial_capital=100000): self.df = df.copy self.strategies = strategies self.capital = initial_capital self.position =0 defrun(self): for i inrange(1,len(self.df)): signals =[strategy(self.df.iloc[:i])for strategy in self.strategies] final_signal =sum(signals)/len(signals) if final_signal >0: self.position = self.capital / self.df['close'].iloc[i] self.capital =0 elif final_signal def multi_exchange_trading_system: exchanges =['binance','kraken','bitfinex'] symbols =['BTC/USDT','ETH/USDT','LTC/USDT'] whileTrue: for exchange, symbol inzip(exchanges, symbols): current_data = get_crypto_data(symbol, exchange) signal = calculate_signal(current_data) if signal ==1: place_buy_order(exchange, symbol) elif signal ==-1: place_sell_order(exchange, symbol) time.sleep(60)来源：散文随风想
