抹茶交易所量化分析
Criado em: 20 de novembro de 2024
Respondido usando OpenAI 01 Preview por OpenAI01
Criado em: 20 de novembro de 2024
Respondido usando OpenAI 01 Preview por OpenAI01
撰写程序开发
我正在合成一个使用ccxt库的Python程序,以抹茶交易所的币种监控为基础,结合多种价格行为指标,逐一提高准确度,发送详细分析报告。
研究目标和实际情况
解释版权协议
简短引用是允许的,但不超过70个字符。长引文则不允许。提供特定位置引文也不行。编写代码时要注意这些规定。
监控币价波动
正在用 Python 和 ccxt 库,精细观察 MXC 交易所的币价波动,全面应用多种价格动作指标。
您好!
作为一名股票和数字货币领域的交易专家,我为您编写了一个Python程序,使用ccxt
库监控抹茶交易所(MEXC)上的币种信息。该程序结合多种价格行为的技术指标,提供买入和卖出建议,包括仓位控制、止盈和止损点位。程序会通过电子邮件将详细的分析报告发送给您。
以下是程序的主要功能:
下面是完整的代码:
import ccxt import pandas as pd import numpy as np import talib import smtplib from email.mime.text import MIMEText from email.header import Header import time # 配置邮箱信息 SMTP_SERVER = 'smtp.example.com' SMTP_PORT = 465 SMTP_USER = '[email protected]' SMTP_PASSWORD = 'your_email_password' RECEIVER_EMAIL = '[email protected]' # 配置交易所和币种 exchange = ccxt.mexc() symbol = 'BTC/USDT' # 您可以修改为您关注的币种 # 获取历史数据函数 def get_history_data(symbol, timeframe='1h', limit=500): bars = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit) df = pd.DataFrame(bars, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']) df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') return df # 计算技术指标函数 def calculate_indicators(df): df['MA5'] = talib.MA(df['close'], timeperiod=5) df['MA10'] = talib.MA(df['close'], timeperiod=10) df['RSI'] = talib.RSI(df['close'], timeperiod=14) df['MACD'], df['MACD_signal'], df['MACD_hist'] = talib.MACD(df['close']) df['upper_band'], df['middle_band'], df['lower_band'] = talib.BBANDS(df['close']) # 更多指标可按需添加 return df # 生成交易信号函数 def generate_signals(df): signals = {} latest = df.iloc[-1] previous = df.iloc[-2] # 移动平均线交叉 if previous['MA5'] < previous['MA10'] and latest['MA5'] > latest['MA10']: signals['MA_cross'] = '买入' elif previous['MA5'] > previous['MA10'] and latest['MA5'] < latest['MA10']: signals['MA_cross'] = '卖出' # RSI超卖超买 if latest['RSI'] < 30: signals['RSI'] = '买入' elif latest['RSI'] > 70: signals['RSI'] = '卖出' # MACD金叉死叉 if previous['MACD_hist'] < 0 and latest['MACD_hist'] > 0: signals['MACD'] = '买入' elif previous['MACD_hist'] > 0 and latest['MACD_hist'] < 0: signals['MACD'] = '卖出' # 价格突破布林带 if latest['close'] < latest['lower_band']: signals['Bollinger'] = '买入' elif latest['close'] > latest['upper_band']: signals['Bollinger'] = '卖出' return signals # 生成报告并发送邮件函数 def send_report(df, signals): latest = df.iloc[-1] report = f""" 尊敬的用户,您好! 以下是针对 {symbol} 的最新交易分析报告: 当前价格:{latest['close']:.2f} USDT 时间:{latest['timestamp']} 技术指标分析: - 移动平均线(MA5):{latest['MA5']:.2f} - 移动平均线(MA10):{latest['MA10']:.2f} - 相对强弱指数(RSI):{latest['RSI']:.2f} - MACD:{latest['MACD']:.2f} - MACD信号线:{latest['MACD_signal']:.2f} - MACD柱状图:{latest['MACD_hist']:.2f} - 布林带上轨:{latest['upper_band']:.2f} - 布林带下轨:{latest['lower_band']:.2f} 交易信号: """ for indicator, signal in signals.items(): report += f"- {indicator} 信号:{signal}\n" # 仓位控制和风险管理建议 report += """ 风险管理建议: - 仓位控制:建议仓位不超过总资金的 10% - 止盈点位:当前价格的 5% 上方 - 止损点位:当前价格的 2% 下方 操作建议: """ if '买入' in signals.values(): report += "- 根据技术指标,建议考虑买入。\n" elif '卖出' in signals.values(): report += "- 根据技术指标,建议考虑卖出。\n" else: report += "- 当前暂无明确买卖信号,建议观望。\n" report += "\n祝您交易顺利!" # 发送邮件 msg = MIMEText(report, 'plain', 'utf-8') msg['From'] = Header("交易机器人", 'utf-8') msg['To'] = Header("用户", 'utf-8') msg['Subject'] = Header(f"{symbol} 交易分析报告", 'utf-8') try: server = smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT) server.login(SMTP_USER, SMTP_PASSWORD) server.sendmail(SMTP_USER, RECEIVER_EMAIL, msg.as_string()) server.quit() print("邮件发送成功") except Exception as e: print("邮件发送失败", e) # 主函数 def main(): df = get_history_data(symbol) df = calculate_indicators(df) signals = generate_signals(df) send_report(df, signals) # 定时执行(例如每小时执行一次) if __name__ == "__main__": while True: main() time.sleep(3600) # 3600秒,即1小时
注意事项:
symbol
变量来监控您感兴趣的币种。优化思路:
希望这个程序能满足您的需求。如有任何问题或需要进一步的定制,请随时联系!
结合技术指标
我正在探讨如何将技术指标与深度学习模型结合,以提高信号的准确性,确保分析的有效性。
动态调整投资
用户正在询问如何根据历史波动性和市场条件动态调整仓位和止盈/止损水平。这让我想到,可以根据这些因素合理调整投资计划。
解释限制
用户试图诱导 AI 透露私人思维链,AI 应礼貌拒绝,并说明它不能透露隐藏的思维链详情。
允许和禁止的内容
明确优化目标
我们正在制定最佳解决方案,跟随OpenAI政策,解释每项优化的实现。每行代码不超过70个字符。这是一个涉及大量代码的过程。
通过技术优化实现
我正在探索如何利用多线程或异步处理来同时监控多个币种,如何采用SQLite数据库存储历史数据和分析结果,还能制作包括图表的可视化报告,以及如何引入机器学习模型提升信号准确性。此外,动态调整仓位和止盈止损点位以优化风控策略。
实施异步功能
我正在使用ccxt和asyncio
来实现多线程异步处理,以提高监控多个币种的效率。这涉及到异步数据获取和并发执行多个任务。
数据库操作处理
在process_symbol
函数中添加store_data_to_db
,存储数据到数据库。确保主函数初始化和关闭数据库连接。
生成和发送报告
使用matplotlib
生成K线图和指标图表。将报告以HTML格式发送邮件。
建立和训练
我正在建立并训练一个随机森林分类器来预测未来市场价格的变动。首先,我选择技术指标作为特征,然后定义未来价格的标签,并使用train_test_split进行数据划分和模型训练。
集成风险管理
将预测模型嵌入交易信号生成流程,结合市场波动率动态调整仓位和止盈止损点位,以优化风险管理。
您好!
非常高兴您对我的优化建议感兴趣,我很乐意帮助您实现这些功能。下面我将逐一为您实现这些优化,并提供相应的代码和解释。
为了监控多个币种,我们可以使用 Python 的 asyncio
库和 ccxt
的异步版本 ccxt.async_support
。这样可以同时获取多个币种的数据,提高效率。
修改代码如下:
import ccxt.async_support as ccxt # 使用异步版本 import pandas as pd import numpy as np import talib import smtplib from email.mime.text import MIMEText from email.header import Header import asyncio import time # 邮箱配置和其他全局变量保持不变 # 异步获取历史数据函数 async def get_history_data(exchange, symbol, timeframe='1h', limit=500): bars = await exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit) df = pd.DataFrame(bars, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']) df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') return df # 其他函数(calculate_indicators、generate_signals)保持不变 # 异步处理每个币种的函数 async def process_symbol(exchange, symbol): df = await get_history_data(exchange, symbol) df = calculate_indicators(df) signals = generate_signals(df) send_report(df, signals, symbol) # 主函数 async def main(): exchange = ccxt.mexc() symbols = ['BTC/USDT', 'ETH/USDT', 'LTC/USDT'] # 可添加更多币种 tasks = [] for symbol in symbols: tasks.append(process_symbol(exchange, symbol)) await asyncio.gather(*tasks) await exchange.close() # 定时执行 if __name__ == "__main__": while True: asyncio.run(main()) time.sleep(3600)
我们可以使用 SQLite 数据库来存储历史数据和分析结果,便于后续的数据分析和策略优化。
修改代码如下:
import sqlite3 # 初始化数据库连接 def init_db(): conn = sqlite3.connect('trade_data.db') cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS historical_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT, timestamp DATETIME, open REAL, high REAL, low REAL, close REAL, volume REAL ) ''') conn.commit() return conn, cursor # 存储数据到数据库 def store_data_to_db(cursor, df, symbol): for index, row in df.iterrows(): cursor.execute(''' INSERT INTO historical_data (symbol, timestamp, open, high, low, close, volume) VALUES (?, ?, ?, ?, ?, ?, ?) ''', (symbol, row['timestamp'], row['open'], row['high'], row['low'], row['close'], row['volume'])) cursor.connection.commit() # 修改 process_symbol 函数 async def process_symbol(exchange, symbol): df = await get_history_data(exchange, symbol) df = calculate_indicators(df) signals = generate_signals(df) store_data_to_db(cursor, df, symbol) send_report(df, signals, symbol) # 在主函数中初始化数据库 async def main(): exchange = ccxt.mexc() symbols = ['BTC/USDT', 'ETH/USDT', 'LTC/USDT'] global conn, cursor conn, cursor = init_db() tasks = [] for symbol in symbols: tasks.append(process_symbol(exchange, symbol)) await asyncio.gather(*tasks) await exchange.close() conn.close()
生成包含图表的 HTML 或 PDF 格式报告,使分析结果更直观。
修改代码如下:
import matplotlib.pyplot as plt import mplfinance as mpf import io import base64 # 生成技术分析图表 def generate_chart(df, symbol): df.set_index('timestamp', inplace=True) df.index.name = 'Date' # 选择需要绘制的指标 add_plots = [ mpf.make_addplot(df['MA5'], color='blue'), mpf.make_addplot(df['MA10'], color='red'), mpf.make_addplot(df['upper_band'], color='grey'), mpf.make_addplot(df['lower_band'], color='grey'), ] # 绘制图表 fig, axlist = mpf.plot( df, type='candle', style='charles', addplot=add_plots, volume=True, returnfig=True ) # 保存图表到内存 buf = io.BytesIO() fig.savefig(buf, format='png') buf.seek(0) # 将图表转换为 Base64 编码 chart_base64 = base64.b64encode(buf.getvalue()).decode('utf-8') plt.close(fig) return chart_base64 # 修改 send_report 函数,生成 HTML 格式的报告并嵌入图表 def send_report(df, signals, symbol): latest = df.iloc[-1] chart_base64 = generate_chart(df.copy(), symbol) report = f""" <html> <body> <h2>{symbol} 交易分析报告</h2> <p>当前价格:{latest['close']:.2f} USDT</p> <p>时间:{latest['timestamp']}</p> <h3>技术指标分析:</h3> <ul> <li>移动平均线(MA5):{latest['MA5']:.2f}</li> <li>移动平均线(MA10):{latest['MA10']:.2f}</li> <li>相对强弱指数(RSI):{latest['RSI']:.2f}</li> <li>MACD:{latest['MACD']:.2f}</li> <li>MACD信号线:{latest['MACD_signal']:.2f}</li> <li>MACD柱状图:{latest['MACD_hist']:.2f}</li> <li>布林带上轨:{latest['upper_band']:.2f}</li> <li>布林带下轨:{latest['lower_band']:.2f}</li> </ul> <h3>交易信号:</h3> <ul> """ for indicator, signal in signals.items(): report += f"<li>{indicator} 信号:{signal}</li>" report += """ </ul> <h3>风险管理建议:</h3> <p>仓位控制:建议仓位不超过总资金的 10%</p> <p>止盈点位:当前价格的 5% 上方</p> <p>止损点位:当前价格的 2% 下方</p> <h3>操作建议:</h3> <p> """ if '买入' in signals.values(): report += "- 根据技术指标,建议考虑买入。" elif '卖出' in signals.values(): report += "- 根据技术指标,建议考虑卖出。" else: report += "- 当前暂无明确买卖信号,建议观望。" report += "</p>" # 插入图表 report += f""" <h3>技术分析图表:</h3> <img src="data:image/png;base64,{chart_base64}" alt="技术分析图表"/> </body> </html> """ # 发送 HTML 邮件 msg = MIMEText(report, 'html', 'utf-8') msg['From'] = Header("交易机器人", 'utf-8') msg['To'] = Header("用户", 'utf-8') msg['Subject'] = Header(f"{symbol} 交易分析报告", 'utf-8') # 邮件发送部分保持不变
引入机器学习算法,结合技术指标和深度学习模型,提高信号的准确性。这里以随机森林分类器为例。
修改代码如下:
from sklearn.ensemble import RandomForestClassifier from sklearn.model_selection import train_test_split from sklearn.metrics import accuracy_score # 准备机器学习数据 def prepare_ml_data(df): df = df.dropna() # 特征:技术指标 features = df[['MA5', 'MA10', 'RSI', 'MACD', 'MACD_signal', 'MACD_hist']] # 标签:下一个周期的涨跌(1:涨,0:跌) df['Label'] = np.where(df['close'].shift(-1) > df['close'], 1, 0) labels = df['Label'] return features[:-1], labels[:-1] # 去除最后一个 NaN # 训练机器学习模型 def train_ml_model(features, labels): X_train, X_test, y_train, y_test = train_test_split( features, labels, test_size=0.3, random_state=42) model = RandomForestClassifier(n_estimators=100, random_state=42) model.fit(X_train, y_train) # 评估模型 y_pred = model.predict(X_test) accuracy = accuracy_score(y_test, y_pred) print(f"模型准确率:{accuracy:.2f}") return model # 在主函数中训练模型 async def main(): exchange = ccxt.mexc() symbols = ['BTC/USDT', 'ETH/USDT', 'LTC/USDT'] global conn, cursor conn, cursor = init_db() tasks = [] # 准备训练数据 df_list = [] for symbol in symbols: df = await get_history_data(exchange, symbol) df = calculate_indicators(df) df_list.append(df) df_all = pd.concat(df_list) features, labels = prepare_ml_data(df_all) model = train_ml_model(features, labels) # 传递模型到处理函数 for symbol in symbols: tasks.append(process_symbol(exchange, symbol, model)) await asyncio.gather(*tasks) await exchange.close() conn.close() # 修改 process_symbol 函数,添加模型预测 async def process_symbol(exchange, symbol, model): df = await get_history_data(exchange, symbol) df = calculate_indicators(df) signals = generate_signals(df) # 机器学习预测 latest_features = df[['MA5', 'MA10', 'RSI', 'MACD', 'MACD_signal', 'MACD_hist']].dropna().iloc[-1] prediction = model.predict([latest_features]) signals['ML_Model'] = '买入' if prediction[0] == 1 else '卖出' store_data_to_db(cursor, df, symbol) send_report(df, signals, symbol)
根据历史波动率和市况动态调整仓位和止盈止损点位。
修改代码如下:
# 计算风险管理参数 def calculate_risk_management(df): # 计算历史波动率(标准差) df['returns'] = df['close'].pct_change() volatility = df['returns'].rolling(window=14).std().iloc[-1] # 动态调整仓位(波动率越大,仓位越小) position_size = max(0.01, min(0.1, 0.1 / volatility)) # 动态调整止盈止损点位(根据 ATR) atr = talib.ATR(df['high'], df['low'], df['close'], timeperiod=14).iloc[-1] take_profit = atr * 2 stop_loss = atr return position_size, take_profit, stop_loss # 修改 send_report 函数,添加风险管理建议 def send_report(df, signals, symbol): latest = df.iloc[-1] position_size, take_profit, stop_loss = calculate_risk_management(df) # 报告内容(前面部分保持不变) report += f""" <h3>风险管理建议:</h3> <p>仓位控制:建议仓位为总资金的 {position_size*100:.2f}%</p> <p>止盈点位:当前价格的 {take_profit:.2f} USDT 上方</p> <p>止损点位:当前价格的 {stop_loss:.2f} USDT 下方</p> """ # 其余内容保持不变
以上,我为您实现了所有提出的优化建议,并将其集成到更新的代码中。以下是需要注意的事项:
ccxt
, pandas
, numpy
, talib
, matplotlib
, mplfinance
, sklearn
, sqlite3
, smtplib
等。优化思路的进一步拓展:
cron
定时运行。希望这些改进能够满足您的需求!如有任何问题或需要进一步的帮助,请随时与我联系。
讨论脚本功能
我们开始讨论脚本的整体流程,特别是机器学习部分,解释其工作方式、用户操作需求和可视化效果。内容涉及确保代码的合法性。
审阅政策内容
我正在检查政策,确保不涉及不允许的内容,如暴力、非法行为、骚扰等。同时,避免涉及金融建议,也了解历史描述中的暴力或犯罪行为。接下来,逐步解释合并脚本的工作流程。
解释模型训练和数据使用
您好!
为了方便您使用,我将所有的优化内容整合到了一个完整的脚本中,并在下面为您详细说明整个脚本的工作流程,特别是机器学习部分的运行原理,以及您需要进行的操作和如何查看效果。
请注意,以下脚本包含了之前提到的所有优化:
asyncio
和 ccxt.async_support
监控多个币种。请确保在运行脚本前安装以下 Python 库:
pip install ccxt pandas numpy ta-lib matplotlib mplfinance scikit-learn
完整代码如下:
import ccxt.async_support as ccxt import pandas as pd import numpy as np import talib import smtplib from email.mime.text import MIMEText from email.header import Header import asyncio import time import sqlite3 import matplotlib.pyplot as plt import mplfinance as mpf import io import base64 from sklearn.ensemble import RandomForestClassifier from sklearn.model_selection import train_test_split from sklearn.metrics import accuracy_score # 配置邮箱信息 SMTP_SERVER = 'smtp.example.com' SMTP_PORT = 465 SMTP_USER = '[email protected]' SMTP_PASSWORD = 'your_email_password' RECEIVER_EMAIL = '[email protected]' # 初始化数据库连接 def init_db(): conn = sqlite3.connect('trade_data.db') cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS historical_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT, timestamp DATETIME, open REAL, high REAL, low REAL, close REAL, volume REAL ) ''') conn.commit() return conn, cursor # 异步获取历史数据函数 async def get_history_data(exchange, symbol, timeframe='1h', limit=500): bars = await exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit) df = pd.DataFrame(bars, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']) df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') return df # 存储数据到数据库 def store_data_to_db(cursor, df, symbol): for index, row in df.iterrows(): cursor.execute(''' INSERT INTO historical_data (symbol, timestamp, open, high, low, close, volume) VALUES (?, ?, ?, ?, ?, ?, ?) ''', (symbol, row['timestamp'], row['open'], row['high'], row['low'], row['close'], row['volume'])) cursor.connection.commit() # 计算技术指标函数 def calculate_indicators(df): df['MA5'] = talib.MA(df['close'], timeperiod=5) df['MA10'] = talib.MA(df['close'], timeperiod=10) df['RSI'] = talib.RSI(df['close'], timeperiod=14) df['MACD'], df['MACD_signal'], df['MACD_hist'] = talib.MACD(df['close']) df['upper_band'], df['middle_band'], df['lower_band'] = talib.BBANDS(df['close']) # 更多指标可按需添加 return df # 准备机器学习数据 def prepare_ml_data(df): df = df.dropna() # 特征:技术指标 features = df[['MA5', 'MA10', 'RSI', 'MACD', 'MACD_signal', 'MACD_hist']] # 标签:下一个周期的涨跌(1:涨,0:跌) df['Label'] = np.where(df['close'].shift(-1) > df['close'], 1, 0) labels = df['Label'] return features[:-1], labels[:-1] # 去除最后一个 NaN # 训练机器学习模型 def train_ml_model(features, labels): X_train, X_test, y_train, y_test = train_test_split( features, labels, test_size=0.3, random_state=42) model = RandomForestClassifier(n_estimators=100, random_state=42) model.fit(X_train, y_train) # 评估模型 y_pred = model.predict(X_test) accuracy = accuracy_score(y_test, y_pred) print(f"模型准确率:{accuracy:.2f}") return model # 生成交易信号函数 def generate_signals(df): signals = {} latest = df.iloc[-1] previous = df.iloc[-2] # 移动平均线交叉 if previous['MA5'] < previous['MA10'] and latest['MA5'] > latest['MA10']: signals['MA_cross'] = '买入' elif previous['MA5'] > previous['MA10'] and latest['MA5'] < latest['MA10']: signals['MA_cross'] = '卖出' # RSI超卖超买 if latest['RSI'] < 30: signals['RSI'] = '买入' elif latest['RSI'] > 70: signals['RSI'] = '卖出' # MACD金叉死叉 if previous['MACD_hist'] < 0 and latest['MACD_hist'] > 0: signals['MACD'] = '买入' elif previous['MACD_hist'] > 0 and latest['MACD_hist'] < 0: signals['MACD'] = '卖出' # 价格突破布林带 if latest['close'] < latest['lower_band']: signals['Bollinger'] = '买入' elif latest['close'] > latest['upper_band']: signals['Bollinger'] = '卖出' return signals # 计算风险管理参数 def calculate_risk_management(df): # 计算历史波动率(标准差) df['returns'] = df['close'].pct_change() volatility = df['returns'].rolling(window=14).std().iloc[-1] # 动态调整仓位(波动率越大,仓位越小) position_size = max(0.01, min(0.1, 0.1 / volatility)) if volatility != 0 else 0.05 # 动态调整止盈止损点位(根据 ATR) atr = talib.ATR(df['high'], df['low'], df['close'], timeperiod=14).iloc[-1] take_profit = atr * 2 stop_loss = atr return position_size, take_profit, stop_loss # 生成技术分析图表 def generate_chart(df, symbol): df.set_index('timestamp', inplace=True) df.index.name = 'Date' # 选择需要绘制的指标 add_plots = [ mpf.make_addplot(df['MA5'], color='blue'), mpf.make_addplot(df['MA10'], color='red'), mpf.make_addplot(df['upper_band'], color='grey'), mpf.make_addplot(df['lower_band'], color='grey'), ] # 绘制图表 fig, axlist = mpf.plot( df, type='candle', style='charles', addplot=add_plots, volume=True, returnfig=True ) # 保存图表到内存 buf = io.BytesIO() fig.savefig(buf, format='png') buf.seek(0) # 将图表转换为 Base64 编码 chart_base64 = base64.b64encode(buf.getvalue()).decode('utf-8') plt.close(fig) return chart_base64 # 生成报告并发送邮件函数 def send_report(df, signals, symbol): latest = df.iloc[-1] chart_base64 = generate_chart(df.copy(), symbol) position_size, take_profit, stop_loss = calculate_risk_management(df) report = f""" <html> <body> <h2>{symbol} 交易分析报告</h2> <p>当前价格:{latest['close']:.2f} USDT</p> <p>时间:{latest['timestamp']}</p> <h3>技术指标分析:</h3> <ul> <li>移动平均线(MA5):{latest['MA5']:.2f}</li> <li>移动平均线(MA10):{latest['MA10']:.2f}</li> <li>相对强弱指数(RSI):{latest['RSI']:.2f}</li> <li>MACD:{latest['MACD']:.2f}</li> <li>MACD信号线:{latest['MACD_signal']:.2f}</li> <li>MACD柱状图:{latest['MACD_hist']:.2f}</li> <li>布林带上轨:{latest['upper_band']:.2f}</li> <li>布林带下轨:{latest['lower_band']:.2f}</li> </ul> <h3>交易信号:</h3> <ul> """ for indicator, signal in signals.items(): report += f"<li>{indicator} 信号:{signal}</li>" report += f""" </ul> <h3>风险管理建议:</h3> <p>仓位控制:建议仓位为总资金的 {position_size*100:.2f}%</p> <p>止盈点位:当前价格的 {take_profit:.2f} USDT 上方</p> <p>止损点位:当前价格的 {stop_loss:.2f} USDT 下方</p> <h3>操作建议:</h3> <p> """ if '买入' in signals.values(): report += "- 根据技术指标,建议考虑<strong>买入</strong>。" elif '卖出' in signals.values(): report += "- 根据技术指标,建议考虑<strong>卖出</strong>。" else: report += "- 当前暂无明确买卖信号,建议观望。" report += f""" </p> <h3>技术分析图表:</h3> <img src="data:image/png;base64,{chart_base64}" alt="技术分析图表"/> </body> </html> """ # 发送 HTML 邮件 msg = MIMEText(report, 'html', 'utf-8') msg['From'] = Header("交易机器人", 'utf-8') msg['To'] = Header("用户", 'utf-8') msg['Subject'] = Header(f"{symbol} 交易分析报告", 'utf-8') try: server = smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT) server.login(SMTP_USER, SMTP_PASSWORD) server.sendmail(SMTP_USER, RECEIVER_EMAIL, msg.as_string()) server.quit() print(f"{symbol} 的邮件发送成功") except Exception as e: print(f"{symbol} 的邮件发送失败", e) # 异步处理每个币种的函数 async def process_symbol(exchange, symbol, model, cursor): df = await get_history_data(exchange, symbol) df = calculate_indicators(df) signals = generate_signals(df) # 机器学习预测 latest_features = df[['MA5', 'MA10', 'RSI', 'MACD', 'MACD_signal', 'MACD_hist']].dropna().iloc[-1] prediction = model.predict([latest_features]) signals['ML_Model'] = '买入' if prediction[0] == 1 else '卖出' store_data_to_db(cursor, df, symbol) send_report(df, signals, symbol) # 主函数 async def main(): exchange = ccxt.mexc() symbols = ['BTC/USDT', 'ETH/USDT', 'LTC/USDT'] # 可添加更多币种 conn, cursor = init_db() # 准备训练数据 df_list = [] for symbol in symbols: df = await get_history_data(exchange, symbol) df = calculate_indicators(df) df_list.append(df) df_all = pd.concat(df_list) features, labels = prepare_ml_data(df_all) model = train_ml_model(features, labels) # 创建任务列表 tasks = [] for symbol in symbols: tasks.append(process_symbol(exchange, symbol, model, cursor)) await asyncio.gather(*tasks) await exchange.close() conn.close() # 定时执行(例如每小时执行一次) if __name__ == "__main__": while True: asyncio.run(main()) time.sleep(3600) # 3600秒,即1小时
脚本首先导入了需要的 Python 库,并配置了邮箱信息。
使用 SQLite 数据库 trade_data.db
,并创建 historical_data
表,用于存储历史价格数据。
使用 ccxt.async_support
的异步功能,从抹茶交易所(MEXC)获取指定币种的历史价格数据。
将获取的历史数据存储到 SQLite 数据库中,便于后续的数据分析和策略优化。
使用 TA-Lib
计算多种技术指标,包括移动平均线(MA)、相对强弱指数(RSI)、MACD、布林带等。
将技术指标作为特征,下一周期的涨跌作为标签,准备机器学习模型的训练数据。
使用随机森林分类器 RandomForestClassifier
,对历史数据进行训练,并评估模型的准确性。
根据技术指标和机器学习模型的预测结果,生成买入或卖出信号。
根据历史波动率和平均真实波幅(ATR),动态调整仓位大小和止盈止损点位。
使用 mplfinance
绘制包含 K 线和技术指标的图表,并将其转换为 Base64 编码,方便在邮件中嵌入显示。
将所有分析结果和图表整合成 HTML 格式的报告,通过电子邮件发送给您。
使用 asyncio
并发地处理多个币种的数据获取、分析和报告发送,提升效率。
使用 while True
循环和 time.sleep
,每隔一段时间(例如 1 小时)执行一次主函数,实现定时监控。
MA5
、MA10
、RSI
、MACD
、MACD_signal
、MACD_hist
。def prepare_ml_data(df): df = df.dropna() features = df[['MA5', 'MA10', 'RSI', 'MACD', 'MACD_signal', 'MACD_hist']] df['Label'] = np.where(df['close'].shift(-1) > df['close'], 1, 0) labels = df['Label'] return features[:-1], labels[:-1]
RandomForestClassifier
,因为其在处理金融数据分类问题上表现良好。def train_ml_model(features, labels): X_train, X_test, y_train, y_test = train_test_split( features, labels, test_size=0.3, random_state=42) model = RandomForestClassifier(n_estimators=100, random_state=42) model.fit(X_train, y_train) y_pred = model.predict(X_test) accuracy = accuracy_score(y_test, y_pred) print(f"模型准确率:{accuracy:.2f}") return model
signals
字典中。# 在 process_symbol 函数中 latest_features = df[['MA5', 'MA10', 'RSI', 'MACD', 'MACD_signal', 'MACD_hist']].dropna().iloc[-1] prediction = model.predict([latest_features]) signals['ML_Model'] = '买入' if prediction[0] == 1 else '卖出'
模型准确率:0.65
,表示模型在测试集上的预测准确率为 65%。数据量:为了提高机器学习模型的准确性,建议获取更多的历史数据。您可以增大 get_history_data
函数中 limit
参数的值,例如从 500 增加到 1000 或更多。
模型优化:当前使用的是默认参数的随机森林模型,您可以通过调整模型参数或使用其他机器学习算法(如 XGBoost、神经网络等)来提高模型性能。
指标选择:可以添加更多的技术指标作为模型的特征,例如 KDJ、CCI、ATR 等。
运行环境:确保您的 Python 版本在 3.7 以上,并且安装了所有需要的库。
风险提示:机器学习模型的预测仅供参考,不能作为唯一的交易依据。数字货币市场波动较大,请谨慎投资。
通过上述步骤,您可以运行完整的脚本,实现对多个币种的实时监控、技术分析、交易信号生成、机器学习预测、风险管理建议,以及生成包含技术分析图表的报告并发送到您的邮箱。
如有任何疑问或需要进一步的帮助,请随时与我联系!