1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
|
import networkx as nx
from loguru import logger
import datetime
import traceback
import json
from data.redis_data_store import RedisDataStorage
class SimpleArbitrageGraph:
# 默认的交易所手续费配置
DEFAULT_FEE_CONFIG = {
'binance': {'buy': 0.001, 'sell': 0.001, "withdraw": 0.001}, # Binance 现货手续费率
'upbit': {'buy': 0.0004, 'sell': 0.0004, "withdraw": 0.001}, # Upbit 现货手续费率
'bithumb': {'buy': 0.002, 'sell': 0.002, "withdraw": 0.001}, # Bithumb 现货手续费率
'gate': {'buy': 0.002, 'sell': 0.002, "withdraw": 0.001},
'okx': {'buy': 0.002, 'sell': 0.002, "withdraw": 0.001},
'mexc': {'buy': 0.002, 'sell': 0.002, "withdraw": 0.001},
'bitget': {'buy': 0.002, 'sell': 0.002, "withdraw": 0.001},
'bybit': {'buy': 0.002, 'sell': 0.002, "withdraw": 0.001},
'xt': {'buy': 0.002, 'sell': 0.002, "withdraw": 0.001},
'lbank': {'buy': 0.002, 'sell': 0.002, "withdraw": 0.001},
'huobi': {'buy': 0.002, 'sell': 0.002, "withdraw": 0.001},
}
def __init__(self, fee_config=None):
"""
初始化套利图构建器
fee_config: 交易所手续费配置字典，如果不传入则使用默认配置
"""
self.fee_config = fee_config if fee_config is not None else self.DEFAULT_FEE_CONFIG
self.graph = nx.MultiDiGraph() # 使用多重有向图
self.opportunity_durations = {} # 用于记录套利机会的持续时间
self.redis_client = RedisDataStorage()
def addOrUpdateEdge(self, from_currency, to_currency, price, exchange, orderbook=None):
"""
添加或更新交易对（边）到图中
from_currency: 起始币种
to_currency: 目标币种
price: 当前交易对的价格
exchange: 交易所名称
orderbook: 订单簿数据，格式为：{
'bids': [(price, amount), ...], # 买单列表，按价格降序
'asks': [(price, amount), ...], # 卖单列表，按价格升序
'timestamp': timestamp # 订单簿时间戳
}
"""
# 获取当前交易所的买入和卖出手续费
fee = self.fee_config[exchange]['buy'] + self.fee_config[exchange]['sell']
# 准备边的属性
edge_data = {
'weight': price,
'fee': fee,
'exchange': exchange,
'key': exchange,
'orderbook': None
}
# 如果提供了orderbook数据，添加到边的属性中并计算统计信息
if orderbook:
bids = orderbook['bids']
asks = orderbook['asks']
# 计算bids的统计信息
try:
bids_total_volume = sum(amount for _, amount in bids)
bids_weighted_price = sum(price * amount for price, amount in bids) / bids_total_volume if bids_total_volume > 0 else 0
except Exception as e:
logger.error(f"Error calculating bids stats: {e}")
logger.error(f"{exchange} {from_currency} - {to_currency} {bids}")
# 计算asks的统计信息
asks_total_volume = sum(amount for _, amount in asks)
asks_weighted_price = sum(price * amount for price, amount in asks) / asks_total_volume if asks_total_volume > 0 else 0
edge_data['orderbook'] = {
'bids': bids,
'asks': asks,
'timestamp': orderbook.get('timestamp', None),
'stats': {
'bids': {
'total_volume': bids_total_volume,
'weighted_price': bids_weighted_price
},
'asks': {
'total_volume': asks_total_volume,
'weighted_price': asks_weighted_price
}
}
}
logger.debug(f"Added orderbook stats info:{edge_data['orderbook']}")
# 如果边已存在，更新它
if self.graph.has_edge(from_currency, to_currency):
for edge in self.graph[from_currency][to_currency].values():
if edge['exchange'] == exchange:
# 更新基本属性
edge['weight'] = price
edge['fee'] = fee
# 确保orderbook字段存在
if 'orderbook' not in edge:
edge['orderbook'] = {}
# 更新orderbook数据
if orderbook:
edge['orderbook'] = edge_data['orderbook']
logger.debug(f"Updated edge from {from_currency} to {to_currency}: price = {price}, fee = {fee}, exchange = {exchange}")
return
# 如果边不存在，添加新边
self.graph.add_edge(from_currency, to_currency, **edge_data)
logger.debug(f"Added edge from {from_currency} to {to_currency}: price = {price}, fee = {fee}, exchange = {exchange}")
def calculate_cross_exchange_arbitrage(self, base_currency, quote_currency, intermediate_currency, amount, cex_a, cex_b):
"""
计算跨交易所套利机会
base_currency: 基础币种（如BTC）
quote_currency: 报价币种（如USDT）
intermediate_currency: 中间币种（如KRW）
amount: 交易金额（以quote_currency计价）
cex_a: 交易所A的名称
cex_b: 交易所B的名称
return: (套利机会详细信息字典, 套利路径描述)
"""
# 检查交易所A中是否存在直接交易对
if not self.graph.has_edge(base_currency, quote_currency):
logger.debug(f"在交易所{cex_a}中未找到{base_currency}/{quote_currency}交易对")
return None, "",""
# 在交易所A中计算可以买入的base_currency数量
base_amount = None
cost_in_a = amount
cost_edge = None
cost_counter_price = None
for edge in self.graph[base_currency][quote_currency].values():
if edge['exchange'] == cex_a:
# 检查orderbook数据
if 'orderbook' in edge and edge['orderbook'] and 'asks' in edge['orderbook'] :
# 获取asks的统计数据
asks_stats = edge['orderbook']['stats']['asks']
# 检查是否有足够的卖单深度
if asks_stats['total_volume'] < amount:
logger.warning(f"订单深度不足：需要{amount} {quote_currency}，但只有{asks_stats['total_volume']} {quote_currency}的深度")
return None, "",""
# 使用加权平均价格计算可买入数量
weighted_price = asks_stats['weighted_price']
cost_counter_price = weighted_price
base_amount = amount / (weighted_price * (1 + edge['fee']))
else:
cost_counter_price = edge['weight']
# 如果没有orderbook数据，使用普通价格
base_amount = amount / (edge['weight'] * (1 + edge['fee']))
cost_edge = edge
logger.debug(f"在{cex_a}中用{amount} {quote_currency}可以买入{base_amount} {base_currency}")
break
if base_amount is None:
logger.debug(f"在交易所{cex_a}中未找到{base_currency}/{quote_currency}交易对")
return None, "",""
# 检查交易所B中的转换路径
try:
# 第一步：计算在交易所B中将base_currency转换为intermediate_currency
base_to_intermediate_edge = None
if not self.graph.has_edge(base_currency, intermediate_currency):
logger.warning(f"在交易所{cex_b}中未找到{base_currency}/{intermediate_currency}交易对")
return None, "",""
for edge in self.graph[base_currency][intermediate_currency].values():
if edge['exchange'] == cex_b:
base_to_intermediate_edge = edge
break
if base_to_intermediate_edge is None:
logger.warning(f"在交易所{cex_b}中未找到{base_currency}/{intermediate_currency}交易对")
return None, "",""
# 第二步：计算在交易所B中将intermediate_currency转换为quote_currency
intermediate_to_quote_edge = None
for edge in self.graph[intermediate_currency][quote_currency].values():
if edge['exchange'] == cex_b:
intermediate_to_quote_edge = edge
break
if intermediate_to_quote_edge is None:
logger.warning(f"在交易所{cex_b}中未找到{intermediate_currency}/{quote_currency}交易对")
return None, "",""
# 打印详细的交易路径信息
logger.debug(f"交易路径详情:")
logger.debug(f"第一步: 在{cex_a}中买入{amount} {base_currency}")
logger.debug(f"第二步: 在{cex_b}中将{base_currency}转换为{intermediate_currency}")
logger.debug(f"第三步: 在{cex_b}中将{intermediate_currency}转换为{quote_currency}")
# 计算在交易所B中的转换后金额
# 第一次转换：base_currency -> intermediate_currency
base_to_intermediate_price = None
if 'orderbook' in base_to_intermediate_edge and base_to_intermediate_edge['orderbook'] and 'asks' in base_to_intermediate_edge['orderbook']:
# 检查bids深度
bids_stats = base_to_intermediate_edge['orderbook']['stats']['bids']
if bids_stats['total_volume'] < base_amount:
logger.warning(f"订单深度不足：需要卖出{base_amount} {base_currency}，但只有{bids_stats['total_volume']} {base_currency}的深度")
return None, "",""
# 使用加权平均价格计算
weighted_price = bids_stats['weighted_price']
base_to_intermediate_price = weighted_price
intermediate_amount = base_amount * weighted_price * (1 - base_to_intermediate_edge['fee'])
else:
base_to_intermediate_price = base_to_intermediate_edge['weight']
intermediate_amount = base_amount * base_to_intermediate_edge['weight'] * (1 - base_to_intermediate_edge['fee'])
intermediate_to_quote_price = None
# 第二步转换：intermediate_currency -> quote_currency
if 'orderbook' in intermediate_to_quote_edge and intermediate_to_quote_edge['orderbook'] and 'asks' in intermediate_to_quote_edge['orderbook']:
# 检查bids深度
bids_stats = intermediate_to_quote_edge['orderbook']['stats']['bids']
if bids_stats['total_volume'] < intermediate_amount:
logger.warning(f"订单深度不足：需要卖出{intermediate_amount} {intermediate_currency}，但只有{bids_stats['total_volume']} {intermediate_currency}的深度")
return None, "",""
# 使用加权平均价格计算
weighted_price = bids_stats['weighted_price']
intermediate_to_quote_price = weighted_price
final_amount = intermediate_amount * weighted_price * (1 - intermediate_to_quote_edge['fee'])
else:
intermediate_to_quote_price = intermediate_to_quote_edge['weight']
final_amount = intermediate_amount * intermediate_to_quote_edge['weight'] * (1 - intermediate_to_quote_edge['fee'])
# 计算利润（以quote_currency计价）
profit = final_amount - cost_in_a
profit_percentage = (profit / cost_in_a) * 100
# 构建详细的套利信息字典
arbitrage_info = {
'currencies': {
'base': base_currency,
'quote': quote_currency,
'intermediate': intermediate_currency
},
'exchanges': {
'buy': cex_a,
'sell': cex_b
},
'amounts': {
'initial_quote': amount,
'base': base_amount,
'intermediate': intermediate_amount,
'final': final_amount
},
'profit': {
'amount': profit,
'percentage': profit_percentage
}
}
from_config = self.redis_client.get_data_by_exchange(cex_a,
base_currency,
"config")
if from_config and 'config' in from_config:
networks_info = json.loads(from_config['config'])
networks_desc = []
for network in networks_info.get('networks', []):
status = []
if network.get('deposit'):
status.append('可充值')
if network.get('withdraw'):
status.append('可提现')
if network.get('fee'):
status.append("费用:"+str(network.get("fee")))
if status:
networks_desc.append(f"{network['network']}({', '.join(status)})")
networks_description = ' | '.join(networks_desc) if networks_desc else '无网络信息'
else:
networks_description = '无网络配置'
# 修改路径描述部分
# 构建路径描述，添加数据有效性检查
def get_orderbook_stats(edge, side):
if edge and 'orderbook' in edge and edge['orderbook'] and \
'stats' in edge['orderbook'] and side in edge['orderbook']['stats']:
stats = edge['orderbook']['stats'][side]
return f"📊深度: {stats['total_volume']:.8f} 均价: {stats['weighted_price']:.8f}"
return "📊深度数据不可用"
path_description = (
f"套利路径:\n"
f"步骤1: {cex_a} {base_currency}/{quote_currency} 🔵买入价格={cost_counter_price:.8f} "
f"买入{base_amount:.8f} {base_currency}(花费{amount} {quote_currency}) "
f"{get_orderbook_stats(cost_edge, 'asks')}\n"
f"步骤2: {cex_b} {base_currency}/{intermediate_currency} 🟢卖出价格={base_to_intermediate_price:.8f} "
f"转换得到{intermediate_amount:.8f}{intermediate_currency} "
f"{get_orderbook_stats(base_to_intermediate_edge, 'bids')}\n"
f"步骤3: {cex_b} {intermediate_currency}/{quote_currency} 🟢卖出价格={intermediate_to_quote_price:.8f} "
f"转换得到{final_amount:.8f}{quote_currency} "
f"{get_orderbook_stats(intermediate_to_quote_edge, 'bids')}\n"
f"💰总结: 利润={profit:.8f}{quote_currency} (🔥{profit_percentage:.2f}%)\n"
f"⏰时间: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
)
# logger.info(path_description)
return arbitrage_info, path_description, networks_description
except Exception as e:
logger.error(f"计算套利路径时发生错误: {str(e)}, cex_a:{cex_a} base:{base_currency} quote:{quote_currency} cex_b:{cex_b} ")
logger.error(f"Error traceback:\n{traceback.format_exc()}")
return None, "",""
def find_all_cross_exchange_arbitrage(self, amount=1.0, min_profit_threshold=0.001,exchange_router_config=None):
"""
自动发现所有可能的跨交易所套利机会
amount: 交易金额（以base_currency计价）
min_profit_threshold: 最小利润阈值，低于此阈值的套利机会将被忽略
exchange_router_config: 交易所路由配置，格式为：{
'from_exchanges': ['binance', 'okx'], # 允许的买入交易所列表
'to_exchanges': ['upbit', 'bithumb'] # 允许的卖出交易所列表
}
return: 所有可行的套利机会列表
"""
opportunities = []
# 获取所有币种和交易所
currencies = list(self.graph.nodes())
exchanges = list(set(edge['exchange'] for _, _, edge in self.graph.edges(data=True)))
# 定义允许的报价币种
allowed_quote_currencies = ['USDT', 'KRW']
# 收集当前有效的套利机会标识
current_opportunities = set()
# 遍历所有可能的币种组合和交易所组合
for base_currency in currencies:
for quote_currency in allowed_quote_currencies:
if base_currency == quote_currency:
continue
# 遍历配置的交易所组合
for cex_a in exchange_router_config['from_exchanges']:
for cex_b in exchange_router_config['to_exchanges']:
if cex_a == cex_b:
continue
# 遍历所有可能的中间币种
# for intermediate_currency in currencies:
for intermediate_currency in ['KRW']:
if intermediate_currency in [base_currency, quote_currency]:
continue
# 计算套利机会
profit, path_description,networks_description = self.calculate_cross_exchange_arbitrage(
base_currency, quote_currency, intermediate_currency,
amount, cex_a, cex_b
)
profit_value=profit['profit']['amount'] if profit and profit['profit']['amount'] else 0.00000000
# 生成套利机会的唯一标识
opp_key = f"{base_currency}-{quote_currency}-{intermediate_currency}-{cex_a}-{cex_b}"
# 如果利润超过阈值，记录套利机会
# if profit_value and (profit_value / (amount * self.graph[base_currency][quote_currency][cex_a]['weight'])) > min_profit_threshold:
# 如果有利润且有路径描述，记录为有效套利机会
show_all_profit = self.redis_client.get_generic_data("config", "show_all_profit")
if (show_all_profit or profit_value > 0) and path_description:
current_time = datetime.datetime.now()
current_opportunities.add(opp_key)
# 更新持续时间记录
if opp_key not in self.opportunity_durations:
self.opportunity_durations[opp_key] = {
'first_seen': current_time,
'last_seen': current_time,
'duration': 0
}
else:
self.opportunity_durations[opp_key]['last_seen'] = current_time
self.opportunity_durations[opp_key]['duration'] = \
(current_time - self.opportunity_durations[opp_key]['first_seen']).total_seconds()
opportunity = {
'base_currency': base_currency,
'quote_currency': quote_currency,
'intermediate_currency': intermediate_currency,
'cex_a': cex_a,
'cex_b': cex_b,
'profit': profit_value,
'duration': self.opportunity_durations[opp_key]['duration'],
'path_description': path_description,
'network_description': networks_description
}
opportunities.append(opportunity)
logger.info(f"发现新的套利机会：{base_currency}->{intermediate_currency}->{quote_currency} | {cex_a}->{cex_b} | 利润：{profit_value:.8f}")
# 清理已不存在的套利机会记录
expired_opportunities = set(self.opportunity_durations.keys()) - current_opportunities
for opp_key in expired_opportunities:
del self.opportunity_durations[opp_key]
logger.debug(f"移除已消失的套利机会记录：{opp_key}")
# 按利润排序
opportunities.sort(key=lambda x: x['profit'], reverse=True)
return opportunities
def format_opportunities_to_markdown(self, opportunities):
"""
将套利机会列表转换为markdown格式的字符串
opportunities: 套利机会列表
return: markdown格式的字符串
"""
if not opportunities:
return []
# 返回单个套利机会的表格行数据
rows = []
for i, opp in enumerate(opportunities, 1):
row = f"| {i} | {opp['base_currency']} | {opp['quote_currency']} | {opp['intermediate_currency']} | "
row += f"{opp['cex_a']} | {opp['cex_b']} | {opp['profit']:.8f} {opp['quote_currency']} | {int(opp['duration'])} s |"
row += f"{opp['path_description'].replace("\n","<br>")} | {opp['network_description']} | "
rows.append(row)
return rows
def analyze_arbitrage_opportunities(self, configs=None):
"""
分析不同配置下的套利机会
configs: 配置列表，每个配置包含 amount（交易金额）和 threshold（利润阈值）
如果不提供，将使用默认配置
return: (是否发现套利机会, 格式化的套利机会描述（markdown格式）)
"""
if configs is None:
configs = {
'trade':[
{'amount': 1.0, 'threshold': 0.001}, # 标准配置
# {'amount': 0.1, 'threshold': 0.005}, # 小额交易配置
# {'amount': 5.0, 'threshold': 0.0005} # 大额交易配置
],
'exchange_router':{
'from_exchanges': ['binance'], # 允许的买入交易所列表
'to_exchanges': ['upbit', 'bithumb'] # 允许的卖出交易所列表
}
}
all_results = []
for config in configs['trade']:
logger.info(f"\n分析配置：交易金额 = {config['amount']} BTC, 利润阈值 = {config['threshold']*100}%")
opportunities = self.find_all_cross_exchange_arbitrage(
amount=config['amount'],
min_profit_threshold=config['threshold'],
exchange_router_config=configs['exchange_router']
)
# 使用format_opportunities_to_markdown方法获取表格行数据
rows = self.format_opportunities_to_markdown(opportunities)
all_results.extend(rows)
# 如果没有找到任何套利机会
if not all_results:
return False, "未发现任何套利机会。"
# 生成表格头部
final_markdown = "| 序号 | 基础币种 | 报价币种 | 中间币种 | 买入Cex | 卖出Cex | 预期利润 | 持续时间 | 详细路径 | 网络信息 |\n"
final_markdown += "|------|----------|----------|----------|---------|---------|----------|------------|------------|---|\n"
# 添加所有套利机会数据
final_markdown += "\n".join(all_results)
return True, final_markdown
|
