11.6 报单时序竞争¶
在交易中，不同服务器上程序发出的请求（报单）到达交易所有先后顺序，高频交
易对这个先后顺序非常敏感。根据日志文件，统计不同服务器上，在相同时
间点相同合约（股票代码）的报单到达交易所的先后比例。
（1） 每台服务器可能有多份日志文件，每份日志文件的文件名由 PID、MachineID 和日期构
成，例如“5169_MachineE_2025-01-10.log”。MachineID 用于区分不同的服务器。
（2） 程序每笔交易会先发送请求（报单）到交易所，然后收到交易所的应答（报单回报）。
所以每份日志文件中有两类日志，“vendorSendOrderImpl”和“notifyOrder”。其中，
“vendorSendOrderImpl”是报单发送日志，“notifyOrder”是报单回报日志，两类日志中
都包含 OrderRef 字段用于标识同一报单。
注意，不同服务器上的 OrderRef 可能重复，但互相间毫无关系，只有同一个机器上的
相同 OrderRef 才表示同一笔订单。
（3） 以“vendorSendOrderImpl”的日志时间作为报单发单时间，InstrumentID 字段代表报单
交易的合约；以“notifyOrder”日志中的 ExchangeTimestamp 字段作为报单到达交易所
的时间（该时间类似 UnixTime），ExchangeTimestamp 值比较小的合约先到达交易所。
（4） 对于不同服务器上发单时间差值在 200ms 内的相同合约报单，可以认为是相同时间
点的相同合约报单。
注意:相同时间点下，同一台服务器上程序可能在同一个合约上连续报单，即同时有多
笔报单。同一服务器相同时间点上自身的相同合约报单比较没有意义。
（5） 实际实盘时存在会存在各种特殊情况，拓展延伸可挖掘更多数据信息。
示例：
输入：多份日志文件
输出：服务器的报单数量（一次完整的请求及回报记为一次报单数量），相同时间点报单数
量，不同服务器间的先后比例
示例日志文件名
3394_MachineD_2025-01-10.log
21963_MachineA_2025-01-10.log
示例日志文件数据条目
[01/10 09:30:00.525277035] [ info ] [ 3394/3601 ] [notifyOrder] InstrumentID:A Direction:0 OffsetFlag:0 HedgeFlag:1 ConnectionSelectionType:1 OrderRef:2 OrderType:0 ConnectionID:1 RealConnectionID:1 ErrorNo:0 ExchangeRef:0 ExchangeTimestamp:22372 OrderStatus:1 OrderLocalID:5
[01/10 09:30:00.525277421] [ info ] [ 3394/3601 ] [notifyOrder] InstrumentID:A Direction:0 OffsetFlag:0 HedgeFlag:1 ConnectionSelectionType:1 OrderRef:2 OrderType:0 ConnectionID:1 RealConnectionID:1 ErrorNo:0 ExchangeRef:0 ExchangeTimestamp:22372 OrderStatus:2 OrderLocalID:5
#include <algorithm>
#include <cmath>
#include <ctime>
#include <filesystem>
#include <fstream>
#include <iomanip>
#include <iostream>
#include <map>
#include <set>
#include <sstream>
#include <string>
#include <vector>
using namespace std;
// Structure to represent a log entry
struct LogEntry {
string machineID;
long long logTimeMs; // Log time in milliseconds
string logType; // "vendorSendOrderImpl" or "notifyOrder"
int orderRef;
string instrumentID;
long long exchangeTimestamp; // Only for notifyOrder
};
// Structure to represent an order
struct Order {
string machineID;
int orderRef;
long long sendTimeMs; // Time from vendorSendOrderImpl
string instrumentID;
long long exchangeTimestamp; // Time from notifyOrder
bool hasNotify; // Whether it has received notifyOrder
};
// Get all .log files from a directory
vector<string> getLogFilesFromDir(const string& dirPath) {
vector<string> logFiles;
try {
namespace fs = std::filesystem;
if (fs::exists(dirPath) && fs::is_directory(dirPath)) {
for (const auto& entry : fs::directory_iterator(dirPath)) {
if (entry.is_regular_file()) {
string filename = entry.path().filename().string();
if (filename.length() >= 4 &&
filename.substr(filename.length() - 4) == ".log") {
string fullPath = entry.path().string();
logFiles.push_back(fullPath);
}
}
}
}
} catch (const exception& e) {
cerr << "Error reading directory " << dirPath << ": " << e.what() << endl;
}
return logFiles;
}
class LogAnalyzer {
private:
map<pair<string, int>, Order> orders; // Key: (machineID, orderRef)
vector<LogEntry> allEntries;
map<string, int> machineOrderCount; // Total orders per machine
public:
// Parse a single log file
void parseLogFile(const string& filename) {
ifstream file(filename);
if (!file.is_open()) {
cerr << "Cannot open file: " << filename << endl;
return;
}
// Extract MachineID from filename (format: PID_MachineID_Date.log)
string machineID = extractMachineID(filename);
if (machineID.empty()) {
cerr << "Invalid filename format: " << filename << endl;
return;
}
string line;
while (getline(file, line)) {
parseLogLine(line, machineID);
}
file.close();
}
// Extract MachineID from filename
string extractMachineID(const string& filename) {
// Expected format: PID_MachineID_Date.log
size_t lastSlash = filename.find_last_of("/\\");
string fname =
(lastSlash != string::npos) ? filename.substr(lastSlash + 1) : filename;
size_t firstUnderscore = fname.find('_');
size_t secondUnderscore = fname.find('_', firstUnderscore + 1);
if (firstUnderscore != string::npos && secondUnderscore != string::npos) {
return fname.substr(firstUnderscore + 1,
secondUnderscore - firstUnderscore - 1);
}
return "";
}
// Parse a single line of log
void parseLogLine(const string& line, const string& machineID) {
// Format: [MM/DD HH:MM:SS.nnnnnnnnn] [ info ] [ PID/TID ] [logType]
// fields...
if (line.find("[vendorSendOrderImpl]") != string::npos) {
parseSendOrder(line, machineID);
} else if (line.find("[notifyOrder]") != string::npos) {
parseNotifyOrder(line, machineID);
}
}
// Parse vendorSendOrderImpl log
void parseSendOrder(const string& line, const string& machineID) {
// Extract time
long long timeMs = extractTimeMs(line);
// Extract OrderRef and InstrumentID
int orderRef = extractIntField(line, "OrderRef:");
string instrumentID = extractStringField(line, "InstrumentID:");
if (orderRef != -1 && !instrumentID.empty()) {
auto key = make_pair(machineID, orderRef);
if (orders.find(key) == orders.end()) {
Order order;
order.machineID = machineID;
order.orderRef = orderRef;
order.sendTimeMs = timeMs;
order.instrumentID = instrumentID;
order.exchangeTimestamp = 0;
order.hasNotify = false;
orders[key] = order;
} else {
orders[key].sendTimeMs = timeMs;
orders[key].instrumentID = instrumentID;
}
}
}
// Parse notifyOrder log
void parseNotifyOrder(const string& line, const string& machineID) {
// Extract OrderRef and ExchangeTimestamp
int orderRef = extractIntField(line, "OrderRef:");
long long exchangeTs = extractLongField(line, "ExchangeTimestamp:");
string instrumentID = extractStringField(line, "InstrumentID:");
if (orderRef != -1 && exchangeTs != -1) {
auto key = make_pair(machineID, orderRef);
if (orders.find(key) == orders.end()) {
Order order;
order.machineID = machineID;
order.orderRef = orderRef;
order.sendTimeMs = 0;
order.instrumentID = instrumentID;
order.exchangeTimestamp = exchangeTs;
order.hasNotify = true;
orders[key] = order;
} else {
orders[key].exchangeTimestamp = exchangeTs;
orders[key].hasNotify = true;
}
}
}
// Extract time in milliseconds from log timestamp
long long extractTimeMs(const string& line) {
size_t startPos = line.find('[');
if (startPos == string::npos) return -1;
startPos = line.find('[', startPos + 1);
if (startPos == string::npos) return -1;
size_t endPos = line.find(']', startPos);
if (endPos == string::npos) return -1;
string timeStr = line.substr(startPos + 1, endPos - startPos - 1);
// Format: MM/DD HH:MM:SS.nnnnnnnnn
int month, day, hour, minute, second, nanos;
sscanf(timeStr.c_str(), "%d/%d %d:%d:%d.%d", &month, &day, &hour, &minute,
&second, &nanos);
long long ms =
(hour * 3600 + minute * 60 + second) * 1000 + nanos / 1000000;
return ms;
}
// Extract integer field
int extractIntField(const string& line, const string& fieldName) {
size_t pos = line.find(fieldName);
if (pos == string::npos) return -1;
pos += fieldName.length();
size_t endPos = line.find(' ', pos);
if (endPos == string::npos) endPos = line.length();
string value = line.substr(pos, endPos - pos);
try {
return stoi(value);
} catch (...) {
return -1;
}
}
// Extract long field
long long extractLongField(const string& line, const string& fieldName) {
size_t pos = line.find(fieldName);
if (pos == string::npos) return -1;
pos += fieldName.length();
size_t endPos = line.find(' ', pos);
if (endPos == string::npos) endPos = line.length();
string value = line.substr(pos, endPos - pos);
try {
return stoll(value);
} catch (...) {
return -1;
}
}
// Extract string field
string extractStringField(const string& line, const string& fieldName) {
size_t pos = line.find(fieldName);
if (pos == string::npos) return "";
pos += fieldName.length();
size_t endPos = line.find(' ', pos);
if (endPos == string::npos) endPos = line.length();
return line.substr(pos, endPos - pos);
}
// Analyze and output statistics
void analyzeAndOutput() {
// Count complete orders per machine
map<string, int> completeOrdersPerMachine;
vector<Order> completeOrders;
for (auto& orderPair : orders) {
Order& order = orderPair.second;
if (order.hasNotify && order.sendTimeMs > 0) {
completeOrdersPerMachine[order.machineID]++;
completeOrders.push_back(order);
}
}
// Output 1: Order count per server
cout << "=== Server Order Count ===" << endl;
for (auto& machPair : completeOrdersPerMachine) {
cout << machPair.first << ": " << machPair.second << " orders" << endl;
}
cout << endl;
// Group orders by instrument first
map<string, vector<Order>> ordersByInstrument;
for (auto& order : completeOrders) {
ordersByInstrument[order.instrumentID].push_back(order);
}
// Find orders at the same time point (within 200ms, different servers)
vector<vector<Order>> sameTimePointGroups;
for (auto& instPair : ordersByInstrument) {
vector<Order>& instOrders = instPair.second;
// Sort by sendTimeMs
sort(instOrders.begin(), instOrders.end(),
[](const Order& a, const Order& b) {
return a.sendTimeMs < b.sendTimeMs;
});
// Use sliding window to find orders within 200ms
for (size_t i = 0; i < instOrders.size(); ++i) {
vector<Order> timeGroup;
timeGroup.push_back(instOrders[i]);
// Find all orders within 200ms window
for (size_t j = i + 1; j < instOrders.size(); ++j) {
if (instOrders[j].sendTimeMs - instOrders[i].sendTimeMs <= 200) {
timeGroup.push_back(instOrders[j]);
} else {
break;
}
}
// Check if there are orders from different machines
if (timeGroup.size() > 1) {
set<string> machines;
for (auto& order : timeGroup) {
machines.insert(order.machineID);
}
// Only keep groups with orders from different machines
if (machines.size() > 1) {
sameTimePointGroups.push_back(timeGroup);
}
}
}
}
// Count unique orders in same time point groups (avoid duplicate counting)
set<pair<string, int>> countedOrders; // (machineID, orderRef)
int sameTimePointCount = 0;
for (auto& group : sameTimePointGroups) {
for (auto& order : group) {
auto key = make_pair(order.machineID, order.orderRef);
if (countedOrders.find(key) == countedOrders.end()) {
countedOrders.insert(key);
sameTimePointCount++;
}
}
}
// Output 2: Same time point orders count
cout << "=== Same Time Point Orders ===" << endl;
cout << "Same time point (within 200ms, same instrument, different "
"servers) order count: "
<< sameTimePointCount << endl;
cout << endl;
// Output 3: Order arrival ratio between servers
cout << "=== Order Arrival Ratio Between Servers ===" << endl;
outputArrivalRatio(sameTimePointGroups);
}
// Output arrival order ratio
void outputArrivalRatio(const vector<vector<Order>>& groups) {
// (machine1, machine2) -> (machine1_first_count, machine2_first_count)
map<pair<string, string>, pair<int, int>> ratios;
for (auto& group : groups) {
if (group.size() < 2) continue;
set<string> machines;
for (auto& order : group) {
machines.insert(order.machineID);
}
if (machines.size() < 2) continue;
// Sort by exchange timestamp to find arrival order
vector<Order> sortedOrders = group;
sort(sortedOrders.begin(), sortedOrders.end(),
[](const Order& a, const Order& b) {
return a.exchangeTimestamp < b.exchangeTimestamp;
});
// For each pair of orders from different machines, count precedence
for (size_t i = 0; i < sortedOrders.size(); ++i) {
for (size_t j = i + 1; j < sortedOrders.size(); ++j) {
string m1 = sortedOrders[i].machineID;
string m2 = sortedOrders[j].machineID;
if (m1 != m2) {
// sortedOrders[i] arrives first (smaller ExchangeTimestamp)
// Normalize machine order (lexicographically smaller first)
string machineA = (m1 < m2) ? m1 : m2;
string machineB = (m1 < m2) ? m2 : m1;
auto key = make_pair(machineA, machineB);
// Check which machine actually arrived first
if (sortedOrders[i].machineID == machineA) {
ratios[key].first++; // machineA arrives first
} else {
ratios[key].second++; // machineB arrives first
}
}
}
}
}
// Output ratios
if (ratios.empty()) {
cout << "No orders from different servers at same time point." << endl;
} else {
for (auto& ratio : ratios) {
int total = ratio.second.first + ratio.second.second;
if (total > 0) {
double ratio1 = (double)ratio.second.first / total * 100.0;
double ratio2 = (double)ratio.second.second / total * 100.0;
cout << ratio.first.first << " vs " << ratio.first.second << ": "
<< ratio.first.first << " " << fixed << setprecision(2) << ratio1
<< "%, " << ratio.first.second << " " << ratio2 << "%" << endl;
}
}
}
}
};
int main(int argc, char* argv[]) {
LogAnalyzer analyzer;
namespace fs = std::filesystem;
vector<string> logFiles;
// If no arguments provided, use default "LogFile" directory
if (argc < 2) {
string defaultDir = "LogFile";
if (fs::exists(defaultDir) && fs::is_directory(defaultDir)) {
cout << "Reading log files from directory: " << defaultDir << endl;
vector<string> files = getLogFilesFromDir(defaultDir);
logFiles.insert(logFiles.end(), files.begin(), files.end());
} else {
// Try current directory
cout << "Reading log files from current directory..." << endl;
vector<string> files = getLogFilesFromDir(".");
logFiles.insert(logFiles.end(), files.begin(), files.end());
}
} else {
// Process command line arguments
for (int i = 1; i < argc; i++) {
string path = argv[i];
if (fs::exists(path)) {
if (fs::is_directory(path)) {
// It's a directory, scan for .log files
cout << "Reading log files from directory: " << path << endl;
vector<string> files = getLogFilesFromDir(path);
logFiles.insert(logFiles.end(), files.begin(), files.end());
} else if (fs::is_regular_file(path)) {
// It's a file, check if it's a .log file
if (path.length() >= 4 && path.substr(path.length() - 4) == ".log") {
logFiles.push_back(path);
} else {
cerr << "Warning: " << path << " is not a .log file, skipping."
<< endl;
}
}
} else {
cerr << "Warning: " << path << " does not exist, skipping." << endl;
}
}
}
if (logFiles.empty()) {
cerr << "Error: No log files found!" << endl;
cerr << "Usage: " << argv[0]
<< " [directory_or_file1] [directory_or_file2] ..." << endl;
cerr << "If no arguments provided, will search in 'LogFile' directory or "
"current directory."
<< endl;
return 1;
}
cout << "Found " << logFiles.size() << " log file(s) to process." << endl;
cout << endl;
// Parse all log files
for (const auto& logFile : logFiles) {
analyzer.parseLogFile(logFile);
}
// Analyze and output results
analyzer.analyzeAndOutput();
return 0;
}
