基于套接字（Sockets）的Twitter情绪分析·进阶篇
（2/3）· 当EA卡在30分钟轮询与socket握手之间，情绪分数早已不是推文本身的问题
不少交易者把社交情绪EA当成黑箱，以为接上API就能自动开仓。实际上MT5与Python之间靠裸套接字传数据，任何一次OnInit里的连接失败都会让整条信号链静默失效，而日志里往往只剩一个超时。
◍ 情绪信号落地成单的逻辑骨架
把推文情绪转成 MT5 下单动作，核心是一段条件分支：情绪值大于 0 走 buy 分支，小于 0 走 sell 分支，等于 0 则不做任何事。这里 stop_loss 用 InpStopLoss 乘以 _Point 动态算出，点值随品种浮动，黄金和外汇直盘的点值差可能差出十几倍，参数不能直接抄。 开仓前必须用 PositionSelect(symbol1) 查重，已持仓就跳过并打印提示，避免同方向重复加仓把风险敞口叠厚。外汇与贵金属波动剧烈，重复开仓可能让回撤超出预期。 Buy 和 Sell 的挂单都带了止损止盈：buy 的止损放在 price - stop_loss，止盈在 price + take_profit；sell 则反向。下单失败不静默，用 GetLastError() 把错误码打出来，方便你直接在 MT5 终端排查。 和 Python 服务端通信靠本地 socket，InitSocket() 里绑定 127.0.0.1:65432、超时 5000 毫秒。若 SocketCreate 返回负值或 Connect 不成功，函数直接返 -1，EA 起不来时先查这个端口有没有被 Python 进程占用。
class="type">class="kw">double stop_loss = InpStopLoss * _Point; if(tweet_sentiment > class="num">0) { class=class="str">"cmt">// Buy if sentiment is positive if(PositionSelect(symbol1)) { Print("Position already open. Skipping buy."); } else { if(trade.Buy(InpLot, symbol1, price, price - stop_loss, price + take_profit)) Print("Buying ", InpLot, " lots of ", symbol1); else Print("Failed to place buy order. Error: ", GetLastError()); } } else if(tweet_sentiment < class="num">0) { class=class="str">"cmt">// Sell if sentiment is negative if(PositionSelect(symbol1)) { Print("Position already open. Skipping sell."); } else { if(trade.Sell(InpLot, symbol1, price, price + stop_loss, price - take_profit)) Print("Selling ", InpLot, " lots of ", symbol1); else Print("Failed to place sell order. Error: ", GetLastError()); } } } else if(StringFind(result, "ERROR,") == class="num">0) { Print("Error received from Python server: ", result); } else { Print("Unexpected response format: ", result); } } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Initialize socket | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">int InitSocket() { class="type">int socket_handle = SocketCreate(); if(socket_handle < class="num">0) { Print("Error creating socket"); class="kw">return -class="num">1; } Print("Socket created successfully."); class=class="str">"cmt">// Connect to Python server class="type">bool isConnected = SocketConnect(socket_handle, "class="num">127.0.class="num">0.1", class="num">65432, class="num">5000); if(!isConnected) { class="type">int error = GetLastError();
「EA 与 Python 的 socket 收发细节」
在 MT5 里把策略判断甩给 Python 跑，核心就是这套 socket 通信函数。InitSocket 建连失败会直接 Print 错误码并 return -1，EA 侧必须判负，否则后面发数据全是空操作。 CommunicateWithPython 先把字符串用 StringToCharArray 转成 uchar 缓冲，再 SocketSend 发出去；返回字节数小于 0 就关 socket 退出。这里用的是 UTF-8 编码，中文指令或特征名不会乱码。 接收端设了 5000 毫秒超时：用 GetTickCount()+timeout 做截止时间，在 do-while 里靠 SocketIsReadable 探可读长度，SocketRead 拉回来再用 CharArrayToString 拼结果。循环同时受 IsStopped() 约束，策略卸载会立刻断连。 辅助函数 CharArrayToString 只是把 uchar 子段拷进 char 数组再转 string，避免一次性转整包带来偏移错乱。实盘跑之前，建议先在本地起一个 Python socket server 回显，看 Print 里 Data sent / received 的字节数对不对得上。
Print("Error connecting to Python server. Error code: ", error); SocketClose(socket_handle); class="kw">return -class="num">1; } Print("Connection to Python server established."); class="kw">return socket_handle; } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Function to send and receive data | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">class="kw">string CommunicateWithPython(class="type">class="kw">string data) { class="type">int socket_handle = InitSocket(); if(socket_handle < class="num">0) class="kw">return ""; class=class="str">"cmt">// Ensure data is encoded in UTF-class="num">8 class="type">uchar send_buffer[]; StringToCharArray(data, send_buffer); class="type">int bytesSent = SocketSend(socket_handle, send_buffer, ArraySize(send_buffer)); if(bytesSent < class="num">0) { Print("Error sending data!"); SocketClose(socket_handle); class="kw">return ""; } Print("Data sent: ", bytesSent); class="type">uint timeout = class="num">5000; class=class="str">"cmt">// class="num">5 seconds timeout class="type">uchar rsp[]; class="type">class="kw">string result; class="type">uint timeout_check = GetTickCount() + timeout; do { class="type">uint len = SocketIsReadable(socket_handle); if(len) { class="type">int rsp_len; rsp_len = SocketRead(socket_handle, rsp, len, timeout); if(rsp_len > class="num">0) { result += CharArrayToString(rsp, class="num">0, rsp_len); } } } while(GetTickCount() < timeout_check && !IsStopped()); SocketClose(socket_handle); if(result == "") { Print("No data received from Python"); class="kw">return ""; } Print("Data received from Python: ", result); class="kw">return result; } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Helper function to convert class="type">uchar array to class="type">class="kw">string | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">class="kw">string CharArrayToString(class="kw">const class="type">uchar &arr[], class="type">int start, class="type">int length) { class="type">class="kw">string result; class="type">char temp[]; ArrayResize(temp, length); ArrayCopy(temp, arr, class="num">0, start, length); result = CharArrayToString(temp); class="kw">return result; } class=class="str">"cmt">//+------------------------------------------------------------------+
用本地套接字把推特情绪喂给 MT5
这套桥接思路的核心是在本机起一个 TCP 服务器，MT5 EA 通过 socket 把交易品种和 Twitter 凭证发过来，Python 侧抓推文做情绪打分再回传。下面的 start_server 绑定 127.0.0.1:65432，只监听 1 个连接，属于典型的本机低延迟方案，不涉及跨网络暴露。 代码里 recv 缓冲区设为 1024 字节，收到后用 utf-8 解码并忽略错误，再按逗号切分。它硬性要求 inputs 长度等于 8，否则直接抛 ValueError——这意味着 EA 端每次必须规整地传满 8 个字段（含 symbol 与各类 token），少一个就走异常处理分支。 analyze_tweets 用 Bearer Token 走 Twitter API，查询串拼成「{symbol} lang:en -is:retweet」，即只取英文非转推。TextBlob 对单条打情绪分后取均值，这个均值就是回传给 EA 的 tweet_sentiment，EA 据此倾向在正值开买、负值开卖。 别把凭证明文当小事 当前实现每次请求都把 API 密钥在 socket 里明文往返。本机回环还好，若哪天改成远程部署，这套写法会直接泄露凭证，生产环境必须上环境变量或鉴权层。 每 30 分钟抓一批最新推文，系统提供的是近似实时的情绪流，但现版只开仓不跟情绪反转调仓。外汇与贵金属波动受多因子驱动，纯推特情绪信号误触发概率不低，上真实账户前建议在 MT5 策略测试器跑至少三个月样本外验证。
def start_server(): """Starts the server that waits for incoming connections.""" server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.bind((&class="macro">#x27;class="num">127.0.class="num">0.1&class="macro">#x27;, class="num">65432)) server_socket.listen(class="num">1) print("Python server started and waiting for connections...") while True: client_socket, addr = server_socket.accept() print(f"Connection from {addr}") try: data = client_socket.recv(class="num">1024) data = data.decode(&class="macro">#x27;utf-class="num">8&class="macro">#x27;, errors=&class="macro">#x27;ignore&class="macro">#x27;) print(f"Received data: {data}") inputs = data.split(&class="macro">#x27;,&class="macro">#x27;) if len(inputs) != class="num">8: raise ValueError("Eight inputs were expected") symbol, twitter_api_key, twitter_api_secret, twitter_access_token, twitter_access_token_secret, twitter_bearer_token, client_id, client_secret = inputs result = process_data(symbol, twitter_bearer_token) result_string = f"{result[&class="macro">#x27;tweet_sentiment&class="macro">#x27;]}" client_socket.sendall(result_string.encode(&class="macro">#x27;utf-class="num">8&class="macro">#x27;)) print(f"Response sent to client: {result_string}") except Exception as e: print(f"Communication error: {e}") error_message = f"ERROR,{str(e)}" client_socket.sendall(error_message.encode(&class="macro">#x27;utf-class="num">8&class="macro">#x27;)) finally: client_socket.shutdown(socket.SHUT_RDWR) client_socket.close() print("Connection closed") def process_data(symbol, bearer_token): """Processes the data obtained from news and tweets.""" result = { "tweet_sentiment": class="num">0} try: result["tweet_sentiment"] = analyze_tweets(bearer_token, symbol) except Exception as e: raise Exception(f"Error processing data: {e}") print(f"Data processed. Result: {result}") class="kw">return result def analyze_tweets(bearer_token, symbol): """Analyzes recent tweets related to the given symbol.""" try: headers = { &class="macro">#x27;Authorization&class="macro">#x27;: f&class="macro">#x27;Bearer {bearer_token}&class="macro">#x27;, } query = f"{symbol} lang:en -is:retweet" # Get the current time and subtract an hour
◍ 用 Twitter 近因情绪给行情打辅助分
这段 Python 片段演示了如何拉取最近 4 小时、剔除当前 10 秒噪点的推文，再做情感极性均值。时间窗设为 end_time 减 10 秒、再往前推 4 小时，转成 RFC3339 的 'Z' 格式，直接喂给 Twitter v2 recent search 接口，max_results 锁 100 条、按相关度排序。 请求回来先判 status_code，不是 200 就抛异常并打印原文；拿到 data 列表后若为空，函数直接 return 0，意味着这段窗口里没有可计算样本。 sentiments 用 TextBlob 对每条 tweet['text'] 算 polarity，空列表同样回 0。最终 average_sentiment = sum/len，值域约在 [-1,1]，负数倾向恐慌、正数倾向乐观。外汇与贵金属受社交情绪扰动明显，这类信号仅作概率参考，实盘仍属高风险。 把平均极性接进 MT5 的自定义指标或 EA，可在欧美盘重叠时段观察：当均值低于 -0.3 且黄金 15 分钟出现长下影，反弹概率可能抬升，但是否入场还得看价位结构。
end_time = class="type">class="kw">datetime.now(timezone.utc) - timedelta(seconds=class="num">10) # Subtract class="num">10 seconds from the current time start_time = end_time - timedelta(hours=class="num">4) # Convert to RFC class="num">3339 (ISO class="num">8601) format with second precision and &class="macro">#x27;Z&class="macro">#x27; at the end start_time_str = start_time.strftime(&class="macro">#x27;%Y-%m-%dT%H:%M:%SZ&class="macro">#x27;) end_time_str = end_time.strftime(&class="macro">#x27;%Y-%m-%dT%H:%M:%SZ&class="macro">#x27;) search_url = f"https:class=class="str">"cmt">//api.twitter.com/class="num">2/tweets/search/recent?query={query}&max_results=class="num">100&start_time={start_time_str}&end_time={end_time_str}&sort_order=relevancy" print(f"Performing tweet search with query: {query}") print(f"Search URL: {search_url}") response = requests.get(search_url, headers=headers) print(f"Response status code: {response.status_code}") print(f"Response text: {response.text}") if response.status_code != class="num">200: raise Exception(f"Error searching tweets: {response.status_code} - {response.text}") tweets = response.json().get(&class="macro">#x27;data&class="macro">#x27;, []) if not tweets: print("No tweets found") class="kw">return class="num">0 sentiments = [TextBlob(tweet[&class="macro">#x27;text&class="macro">#x27;]).sentiment.polarity for tweet in tweets] if not sentiments: print("No sentiments found") class="kw">return class="num">0 average_sentiment = sum(sentiments) / len(sentiments) class="kw">return average_sentiment except Exception as e: print(f"Error: {e}") raise Exception(f"Error analyzing tweets: {e}")