探索加密货币跨交易所套利-用Python构建监控工具

最近在研究加密货币市场时，回想起之前写过一篇用Python获取数字货币行情的文章，当时有读者留言希望我深入讲讲跨交易所的套利机会。起初我觉得，随着参与这类策略的人越来越多，套利空间肯定被压缩得差不多了，就没太上心。但仔细想想，加密货币市场变化多端，说不准还真藏着不少机会呢，索性就趁这个机会好好研究一番，也把过程分享给大家。

在金融投资领域，套利一直是个热门话题。简单来说，套利就是利用不同市场或不同形式的资产之间的价格差异，通过低买高卖来获取利润。跨交易所套利，作为套利策略中的一种，在传统金融市场和加密货币市场都备受关注。在传统金融市场，像股票、期货等交易中，跨交易所套利已经发展得相对成熟，有不少专业机构和投资者参与其中。但加密货币市场可就大不一样了，它24小时不间断交易、交易规则多样，而且各个交易所之间的价格差异也比较常见，这就给跨交易所套利带来了独特的机遇和挑战。

对于加密货币跨交易所套利，可能很多刚接触的朋友还不太清楚。简单来讲，就是在不同的加密货币交易所之间，寻找同一加密货币的价格差异，通过在价格低的交易所买入，在价格高的交易所卖出，等价格回归平衡时平仓，赚取中间的价差收益。比如说，当BTC在Binance的价格是90000美元，而在OKX的价格是90200美元时，理论上就可以在Binance低价买入BTC，然后在OKX高价卖出，从而获得200美元的价差收益。当然，实际操作可没这么简单。

价差套利的类型有很多，像期现套利、跨期套利、跨品种套利等。这些套利策略中，有些风险较低，而有些则可能面临较大风险。比如说，不同品种之间的相关性可能会失效，或者在短期内出现大幅波动，超出了投资者的风险承受能力。今天咱们重点探讨的是加密货币跨交易所套利，并且尝试开发一个工具，来监控不同交易所间相同币种的实时价格，从而发现潜在的套利机会。不过事先声明，这只是我的一次尝试，金融交易风险多多，大家要是想实际操作，可得谨慎再谨慎。

在动手开发这个工具之前，咱们得先理清楚实现的步骤。大致可以分为这么几步：首先，要筛选出跨交易所间能够配对的品种，并且过滤掉或者修正一些错配的情况；接着，监听这些筛选出来的品种的最新价格，计算它们之间的配对价差；最后，从实际应用的角度分析，怎么把监控得到的结果运用到真实的交易当中。下面，咱们就一步步来实现。

配对品种：寻找跨交易所的“孪生”交易对

要实现跨交易所套利，第一步就是找出不同交易所之间能够配对的交易对。这可不像看起来那么简单，里面的门道还真不少。

不同交易所的交易对千差万别，怎么识别哪些可以配对呢？我的方法是通过代码拉取两个交易所的交易对，然后按照计价货币和基准货币来进行配对。就拿BTC/USDT这个交易对来说，计价货币是USDT，基准货币就是BTC。在配对的时候，还得明确品种类型，不然很容易出现不符合预期的配对。比如说，你本来希望配对交易所A和B的永续合约，结果却配成了A交易所的现货和B交易所的永续合约，这可就麻烦大了。

在中心化交易所（CEX）里，品种类型有主类型和子类型的区分。主类型一般有现货spot、永续合约swap、交割合约future、期权option这些。而子类型对于合约来说，不管是永续还是交割，又分为正向合约（U本位）、反向合约（币本位）。这里面期权比较特殊，和其他交易品种差异较大，咱们这次就先不考虑它了。如果对这部分内容不太熟悉，理解起来可能有点困难，要是画张关系图，说不定一下子就清楚了。不过我这人有点懒，就不给大家画了，感兴趣的朋友可以自己动手画一画。

我希望实现一个匹配函数，定义大概是这样的：

defload_pairs(    exchange_a,  # 交易所a实例    exchange_b,  # 交易所b实例    type_a = "spot",  # 品种a主类型    subtype_a = None,  # 品种a子类型    type_b = "spot",  # 品种b主类型    subtype_b = None  # 品种b子类型)

下面就是详细的实现代码：

import osimport ccxtparams = {'enableRateLimit': True,}defload_pairs(    exchange_a,    exchange_b,    type_a = "spot",    subtype_a = None,    type_b = "spot",    subtype_b = None,):    exchange_a.load_markets()    exchange_b.load_markets()    markets_a = {        (m['base'], m['quote']): m['symbol']for m in exchange_a.markets.values()if m['type'] == type_a and (subtype_a isNoneor m[subtype_a])    }    markets_b = {        (m['base'], m['quote']): m['symbol']for m in exchange_b.markets.values()if m['type'] == type_b and (subtype_b isNoneor [subtype_b])    }    pair_keys = set(markets_a.keys()).intersection(set(markets_b.keys()))return [        {'base': base,'quote': quote,'symbol_a': markets_a[(base, quote)],'symbol_b': markets_b[(base, quote)],        }for base, quote in pair_keys    ]

为了让大家更清楚，我给几个示例代码。比如说，想把Binance的现货和OKX的现货交易对关联起来，可以这样写：

binance = ccxt.binance(params)okx = ccxt.okx(params)# Binance现货对OKX现货pairs = load_pairs(binance, okx, type_a='spot', type_b='spot')

要是想关联Binance的现货和OKX的正向永续合约，代码就变成这样：

# Binance现货对OKX正向永续合约pairs = load_pairs(binance, okx, type_a='spot', type_b='swap', subtype_b='linear')

其实同一个交易所的不同类型品种也是可以配对的。像下面这样，就是把OKX的现货和OKX的正向永续合约进行配对：

# 初始化交易所对象okx = ccxt.okx(params)# OKX现货对OKX正向永续合约pairs = load_pairs(okx, okx, type_a='spot', type_b='swap', subtype_b='linear')

但就算按照上面的规则来配对，还是会出现错配的情况。这是为什么呢？原来，不同交易所可能会出现同名不同币的现象。就拿NEIRO来说，在OKX和Bybit上就是不同的币种，而且OKX上有两个NEIRO，其中和Bybit配对的是NEIROETHUSDT合约。具体是什么原因，这里就不细说了。

为了避免这种情况带来的麻烦，我们可以把价差异常的配对找出来，人工确认一下原因。下面这个函数就是用来检测价差异常的：

defdetect_abnormal_pairs(exchange_a, exchange_b, pairs, threshold = 0.05):"""    用于检测价差异常的函数    参数说明：    :param matched_pairs: load_pairs函数返回的匹配交易对列表    :param threshold: 视为异常的价格差异比例（0.05表示5%）    返回结构：    {        'base': 基准货币,        'quote': 计价货币,       'symbol_a': 交易所A交易对,       'symbol_b': 交易所B交易对,        'price_a': 原始价格A,        'price_b': 原始价格B,       'spread_pct': 价差比例,        'is_abnormal': 是否异常    }    """    normal_pairs = []    abonormal_pairs = []for pair in pairs:try:# 获取最新行情数据（单次尝试）            ticker_a = exchange_a.fetch_ticker(pair['symbol_a'])            ticker_b = exchange_b.fetch_ticker(pair['symbol_b'])# 获取最后成交价            price_a = ticker_a.get('last')            price_b = ticker_b.get('last')# 跳过无效价格ifNonein [price_a, price_b]:                print(f"价格缺失: {pair['symbol_a']}/{pair['symbol_b']}")continue# 计算价差比例（基于较小价格）            min_price = min(price_a, price_b)            spread_pct = abs(price_a - price_b) / min_price# 构建结果对象            result = {               **pair,'price_a': price_a,'price_b': price_b,'spread_pct': spread_pct,'is_abnormal': spread_pct > threshold            }if result['is_abnormal']:                abonormal_pairs.append(result)else:                normal_pairs.append(pair)except Exception as e:            print(f"处理交易对 {pair} 时发生错误: {str(e)}")return abonormal_pairs, normal_pairs

为了保险起见，异常的阈值可以设置得小一点，0.05就比较合适。不过，期现的价差有时候可能会大于这个值，这也是需要注意的地方。

下面的代码可以把匹配的交易对保存到csv文件里：

binance = ccxt.binance(params)okx = ccxt.okx(params)pairs = load_pairs(binance, okx,'spot', 'future')abnormal_pairs, normal_pairs = detect_abnormal_pairs(binance, okx, pairs, 0.05)pd.DataFrame(normal_pairs).to_csv("normal_pairs.csv")pd.DataFrame(abnormal_pairs).to_csv("abnormal_pairs.csv")

得到这两个csv文件之后，建议大家手动检查一下里面的配对情况，确保没问题。

还有个容易被忽略的问题，就算是相同币种，这个脚本也有可能漏掉它们。比如说PEPE永续合约，在OKX和Binance上的价格就不一样，因为Binance对PEPE进行了1000倍的放大，交易对名为1000PEPEUSDT，基准货币变成了1000PEPE，而不是PEPE。这种特殊的逻辑我就不在这里实现了，要是有朋友感兴趣，可以自己去研究研究。一般来说，像这种情况大多出现在一些meme币上。

当然了，如果大家不是想全量监控，而是心里清楚自己想监控哪些配对交易对，那就不用进行全量匹配了。

实时监控价差：紧盯价格波动，捕捉套利时机

完成交易对的配对之后，接下来就是要实时监控它们之间的价差了。为了实现实时监控，我借助ccxt.pro里的watch_ticker方法，通过实时监听最新成交价来计算价差。

一开始，我想的是用顺序监控的方式，代码大概是这样：

ticker_a = await exchange_a.watch_ticker(symbol_a)ticker_b = await exchange_b.watch_ticker(symbol_b)spread_pct = abs(ticker_a['last'] - ticker['last'])/min(ticker_b['last'])

但这样做有个问题，就是两个交易所的价格监听存在等待依赖，实时性不太好。为了更及时地获取价格信息，提高监听性能，我决定用watch_tickers方法来批量监听交易所a和b的所有symbol，只要监听到新的价格，就立刻计算价差。

这个实现过程稍微有点复杂，我就不详细展开说了，直接看代码吧：

import osimport asyncioimport ccxt.pro as ccxtproimport pandas as pdimport tracebackfrom typing import Dict, Tuplefrom collections import defaultdictparams = {'enableRateLimit': True,# 'proxies': {#     'http': os.getenv('http_proxy'),#     'https': os.getenv('https_proxy'),# },# 'aiohttp_proxy': os.getenv('http_proxy'),# 'ws_proxy': os.getenv('http_proxy')}classMonitor:def__init__(self, exchange_a, exchange_b, pairs):        self.exchange_a = exchange_a        self.exchange_b = exchange_b# 构建symbol映射关系        self.symbol_map = self._build_symbol_map(pairs)        self.pair_data: Dict[Tuple[str, str], dict] = {}        self.monitor_tasks = []        self.running = Falsedef_build_symbol_map(self, pairs):"""构建symbol到配对关系的快速映射"""        symbol_map = defaultdict(dict)for pair in pairs:            key = (pair['base'], pair['quote'])            symbol_map['a'][pair['symbol_a']] = {'index': 'a', 'pair_key': key}            symbol_map['b'][pair['symbol_b']] = {'index': 'b', 'pair_key': key}return symbol_mapasyncdefmonitor(self, exchange, index: str):"""        统一监控方法        :param exchange: 交易所实例        :param index: 来源索引 ('a'或'b')        """        symbols = [s for s in self.symbol_map[index]]while self.running:try:                tickers = await exchange.watch_tickers(symbols)await self.process_tickers(tickers, index)except Exception as e:                traceback.print_exc()                print(f"监控异常({index}): {str(e)}")await asyncio.sleep(5)asyncdefprocess_tickers(self, tickers, index):"""处理批量ticker数据"""for symbol, ticker in tickers.items():if symbol notin self.symbol_map[index]:continue            pair_map = self.symbol_map[index][symbol]            pair_key = pair_map['pair_key']# 初始化数据结构if pair_key notin self.pair_data:                self.pair_data[pair_key] = {'price_a': None,'price_b': None,'spread': None                }            price_field = f'price_{index}'            self.pair_data[pair_key][price_field] = ticker['last']# 立即计算价差await self.calculate_spread(pair_key)asyncdefcalculate_spread(self, pair_key):"""带校验的价差计算"""        data = self.pair_data[pair_key]try:if data['price_a'] and data['price_b']:                min_price = min(data['price_a'], data['price_b'])                spread_pct = abs(data['price_a'] - data['price_b']) / min_price                data['spread_pct'] = spread_pct# 触发报警的价差阈值if spread > 0.01: await self.trigger_arbitrage(pair_key)except (TypeError, ZeroDivisionError) as e:            print(f"价差计算错误 {pair_key}: {str(e)}")asyncdeftrigger_arbitrage(self, pair_key):"""触发套利逻辑"""        data = self.pair_data[pair_key]        print(f"套利机会! {pair_key} 价差: {data['spread']:.2%}")# 此处添加实际交易逻辑defstart(self):"""启动监控任务"""        self.running = True        self.monitor_tasks = [            asyncio.create_task(self.monitor(self.exchange_a, 'a')),            asyncio.create_task(self.monitor(self.exchange_b, 'b'))        ]asyncdefstop(self):"""优雅关闭"""        self.running = Falseawait asyncio.gather(*self.monitor_tasks, return_exceptions=True)await self.exchange_a.close()await self.exchange_b.close()

在这个Monitor类里，只要监听到一个新的价格，就会重新计算spread价差，并且评估是否触发报警，或者进入到后续的交易验证环节。

下面的代码会导入前面保存的normal_pairs.csv中的交易对，然后监控价差：

asyncdefmain(exchange_a, exchange_b, pairs):await exchange_a.load_markets()await exchange_b.load_markets()    monitor = Monitor(exchange_a, exchange_b, pairs)    monitor.start()try:whileTrue:await asyncio.sleep(1)# 可在此处可添加其他逻辑：except KeyboardInterrupt:await monitor.stop()        print("监控已停止")if __name__ == "__main__":try:        pairs_df = pd.read_csv("normal_pairs.csv")        pairs_dif = pairs_df[pairs_df['quote'] == "USDT"]        required_cols = ['base', 'quote','symbol_a','symbol_b']ifnot all(col in pairs_df.columns for col in required_cols):raise ValueError("CSV文件缺少必要列")        pairs = pairs_df[required_cols].to_dict('records')except Exception as e:        print(f"配置加载失败: {str(e)}")        exit(1)#