如何解决Deal类业务逻辑中的IF嵌套地狱问题？

架构老林

2026-5-13

重构Deal状态处理代码：摆脱IF嵌套地狱

首先，你的核心问题是把状态判断、趋势/交叉条件、业务操作三者耦合在了嵌套的IF里，导致代码可读性差、新增规则成本高。下面我会结合你提到的Strategy和责任链模式，再补充一个更易上手的规则表方案，帮你彻底重构这段代码。

问题回顾

你有一个Deal类，包含New、Bought、Finished状态，以及动态变化的trend和crossing属性，需要根据不同场景（开仓、加仓、止盈）的条件匹配执行对应操作。现有代码陷入了IF嵌套地狱：

async def check_deal(deal):
    if deal.is_new():
        if deal.current_trend == UP_TREND and CROSSED_TOP_GREEN in deal.crossing:
            return await deal.create_order(CROSSED_TOP_GREEN)
        if deal.current_trend == FLAT_TREND and CROSSED_BOTTOM_ORANGE in deal.crossing:
            return await deal.create_order(CROSSED_BOTTOM_ORANGE)
        if deal.current_trend == DOWN_TREND and CROSSED_BOTTOM in deal.crossing:
            return await deal.create_order(CROSSED_BOTTOM)
    if deal.is_bought() and not deal.candle_has_order:
        if deal.current_trend == UP_TREND and CROSSED_BOTTOM_GREEN in deal.crossing:
            return await deal.create_order(CROSSED_BOTTOM_GREEN)
        else:
            return await deal.update_deal()
        if deal.current_trend == DOWN_TREND and CROSSED_BOTTOM in deal.crossing:
            return await deal.create_order(CROSSED_BOTTOM)
        else:
            return await deal.update_deal()
    else:
        return await deal.update_deal()
    if deal.is_profitable() and len(deal.orders) == 1 :
        if deal.current_trend == UP_TREND and CROSSED_TOP in deal.crossing:
            return await deal.trailing_enable(CROSSED_TOP)
        if deal.current_trend == FLAT_TREND and CROSSED_TOP_GREEN in deal.crossing:
            return await deal.trailing_enable(CROSSED_TOP_GREEN)
        if deal.current_trend == DOWN_TREND and CROSSED_BASIS in deal.crossing:
            return await deal.trailing_enable(CROSSED_BASIS)
    if deal.is_profitable() and len(deal.orders) > 1:
        if deal.current_trend == UP_TREND and CROSSED_BASIS in deal.crossing:
            return await deal.trailing_enable(CROSSED_BASIS)
        if deal.current_trend == DOWN_TREND and CROSSED_BOTTOM_GREEN in deal.crossing:
            return await deal.trailing_enable(CROSSED_BOTTOM_GREEN)

核心规则

current_crossing和current_trend为动态变量
开仓：Deal.is_new() + 趋势&交叉匹配
加仓：Deal.is_bought()且无订单 + 趋势&交叉匹配
止盈：Deal盈利 + 订单数量符合 + 趋势&交叉匹配
加仓场景未匹配规则时，执行update_deal()

重构方案

方案1：规则表驱动（最易实现，推荐新手）

把所有规则抽象成条件-映射-动作的三元组，遍历规则表找到第一个匹配项执行，彻底消除嵌套IF。这种方式代码量少，新增规则只需要加条目，无需修改核心逻辑。

# 先定义场景判断函数，把状态/数量等条件封装成独立函数
def is_open_scenario(deal):
    return deal.is_new()

def is_add_position_scenario(deal):
    return deal.is_bought() and not deal.candle_has_order

def is_take_profit_single(deal):
    return deal.is_profitable() and len(deal.orders) == 1

def is_take_profit_multi(deal):
    return deal.is_profitable() and len(deal.orders) > 1

# 规则列表：每个条目是(场景判断, 趋势-交叉映射, 执行动作)
DEAL_RULES = [
    # 开仓规则
    (
        is_open_scenario,
        {
            (UP_TREND, CROSSED_TOP_GREEN): "create_order",
            (FLAT_TREND, CROSSED_BOTTOM_ORANGE): "create_order",
            (DOWN_TREND, CROSSED_BOTTOM): "create_order",
        },
    ),
    # 加仓规则
    (
        is_add_position_scenario,
        {
            (UP_TREND, CROSSED_BOTTOM_GREEN): "create_order",
            (DOWN_TREND, CROSSED_BOTTOM): "create_order",
        },
    ),
    # 止盈（单订单）
    (
        is_take_profit_single,
        {
            (UP_TREND, CROSSED_TOP): "trailing_enable",
            (FLAT_TREND, CROSSED_TOP_GREEN): "trailing_enable",
            (DOWN_TREND, CROSSED_BASIS): "trailing_enable",
        },
    ),
    # 止盈（多订单）
    (
        is_take_profit_multi,
        {
            (UP_TREND, CROSSED_BASIS): "trailing_enable",
            (DOWN_TREND, CROSSED_BOTTOM_GREEN): "trailing_enable",
        },
    ),
]

async def check_deal(deal):
    # 先处理加仓场景的特殊情况：未匹配规则直接执行update_deal
    if is_add_position_scenario(deal):
        matched = False
        for condition, trend_cross_map in DEAL_RULES:
            if condition(deal):
                trigger = next((c for c in deal.crossing if (deal.current_trend, c) in trend_cross_map), None)
                if trigger:
                    matched = True
                    action = getattr(deal, trend_cross_map[(deal.current_trend, trigger)])
                    return await action(trigger)
        if not matched:
            return await deal.update_deal()
    
    # 遍历所有规则找匹配项
    for condition, trend_cross_map in DEAL_RULES:
        if condition(deal):
            # 找到当前crossing中符合规则的触发项
            trigger = next((c for c in deal.crossing if (deal.current_trend, c) in trend_cross_map), None)
            if trigger:
                action = getattr(deal, trend_cross_map[(deal.current_trend, trigger)])
                return await action(trigger)
    
    # 默认执行update_deal
    return await deal.update_deal()

方案2：Strategy模式（面向对象风格）

把每个场景的处理逻辑封装成独立的策略类，根据Deal的状态选择对应策略执行。这种方式更符合OOP思想，便于单元测试和后续扩展新状态。

from abc import ABC, abstractmethod

class DealStrategy(ABC):
    """策略基类"""
    @abstractmethod
    async def execute(self, deal):
        # 返回None表示该策略不处理，交给下一个策略
        pass

class NewDealStrategy(DealStrategy):
    """开仓场景策略"""
    async def execute(self, deal):
        if not deal.is_new():
            return None
        
        trend_cross_map = {
            (UP_TREND, CROSSED_TOP_GREEN): deal.create_order,
            (FLAT_TREND, CROSSED_BOTTOM_ORANGE): deal.create_order,
            (DOWN_TREND, CROSSED_BOTTOM): deal.create_order,
        }
        trigger = next((c for c in deal.crossing if (deal.current_trend, c) in trend_cross_map), None)
        if trigger:
            return await deal.create_order(trigger)
        return None

class BoughtDealStrategy(DealStrategy):
    """加仓场景策略"""
    async def execute(self, deal):
        if not (deal.is_bought() and not deal.candle_has_order):
            return None
        
        trend_cross_map = {
            (UP_TREND, CROSSED_BOTTOM_GREEN): deal.create_order,
            (DOWN_TREND, CROSSED_BOTTOM): deal.create_order,
        }
        trigger = next((c for c in deal.crossing if (deal.current_trend, c) in trend_cross_map), None)
        if trigger:
            return await deal.create_order(trigger)
        # 加仓场景未匹配规则，直接执行update_deal
        return await deal.update_deal()

class ProfitableSingleStrategy(DealStrategy):
    """止盈（单订单）策略"""
    async def execute(self, deal):
        if not (deal.is_profitable() and len(deal.orders) == 1):
            return None
        
        trend_cross_map = {
            (UP_TREND, CROSSED_TOP): deal.trailing_enable,
            (FLAT_TREND, CROSSED_TOP_GREEN): deal.trailing_enable,
            (DOWN_TREND, CROSSED_BASIS): deal.trailing_enable,
        }
        trigger = next((c for c in deal.crossing if (deal.current_trend, c) in trend_cross_map), None)
        if trigger:
            return await deal.trailing_enable(trigger)
        return None

class ProfitableMultiStrategy(DealStrategy):
    """止盈（多订单）策略"""
    async def execute(self, deal):
        if not (deal.is_profitable() and len(deal.orders) > 1):
            return None
        
        trend_cross_map = {
            (UP_TREND, CROSSED_BASIS): deal.trailing_enable,
            (DOWN_TREND, CROSSED_BOTTOM_GREEN): deal.trailing_enable,
        }
        trigger = next((c for c in deal.crossing if (deal.current_trend, c) in trend_cross_map), None)
        if trigger:
            return await deal.trailing_enable(trigger)
        return None

async def check_deal(deal):
    # 按优先级执行策略，第一个处理的策略返回结果
    strategies = [
        NewDealStrategy(),
        BoughtDealStrategy(),
        ProfitableSingleStrategy(),
        ProfitableMultiStrategy()
    ]
    for strategy in strategies:
        result = await strategy.execute(deal)
        if result is not None:
            return result
    # 默认执行update_deal
    return await deal.update_deal()

方案3：责任链模式（规则优先级明确时使用）

把每个规则处理逻辑封装成链中的节点，Deal对象依次经过每个节点，直到某个节点处理该请求。这种方式适合规则有明确优先级，且后续需要动态调整规则顺序的场景。

from abc import ABC, abstractmethod

class Handler(ABC):
    """责任链节点基类"""
    def __init__(self, next_handler=None):
        self.next_handler = next_handler
    
    @abstractmethod
    async def handle(self, deal):
        pass

class NewDealHandler(Handler):
    """开仓节点"""
    async def handle(self, deal):
        if deal.is_new():
            trend_cross_map = {
                (UP_TREND, CROSSED_TOP_GREEN): deal.create_order,
                (FLAT_TREND, CROSSED_BOTTOM_ORANGE): deal.create_order,
                (DOWN_TREND, CROSSED_BOTTOM): deal.create_order,
            }
            trigger = next((c for c in deal.crossing if (deal.current_trend, c) in trend_cross_map), None)
            if trigger:
                return await deal.create_order(trigger)
        # 不处理则传递给下一个节点
        if self.next_handler:
            return await self.next_handler.handle(deal)
        return await deal.update_deal()

class BoughtDealHandler(Handler):
    """加仓节点"""
    async def handle(self, deal):
        if deal.is_bought() and not deal.candle_has_order:
            trend_cross_map = {
                (UP_TREND, CROSSED_BOTTOM_GREEN): deal.create_order,
                (DOWN_TREND, CROSSED_BOTTOM): deal.create_order,
            }
            trigger = next((c for c in deal.crossing if (deal.current_trend, c) in trend_cross_map), None)
            if trigger:
                return await deal.create_order(trigger)
            # 未匹配规则直接执行update_deal，不再传递
            return await deal.update_deal()
        if self.next_handler:
            return await self.next_handler.handle(deal)
        return await deal.update_deal()

class ProfitableSingleHandler(Handler):
    """止盈（单订单）节点"""
    async def handle(self, deal):
        if deal.is_profitable() and len(deal.orders) == 1:
            trend_cross_map = {
                (UP_TREND, CROSSED_TOP): deal.trailing_enable,
                (FLAT_TREND, CROSSED_TOP_GREEN): deal.trailing_enable,
                (DOWN_TREND, CROSSED_BASIS): deal.trailing_enable,
            }
            trigger = next((c for c in deal.crossing if (deal.current_trend, c) in trend_cross_map), None)
            if trigger:
                return await deal.trailing_enable(trigger)
        if self.next_handler:
            return await self.next_handler.handle(deal)
        return await deal.update_deal()

class ProfitableMultiHandler(Handler):
    """止盈（多订单）节点"""
    async def handle(self, deal):
        if deal.is_profitable() and len(deal.orders) > 1:
            trend_cross_map = {
                (UP_TREND, CROSSED_BASIS): deal.trailing_enable,
                (DOWN_TREND, CROSSED_BOTTOM_GREEN): deal.trailing_enable,
            }
            trigger = next((c for c in deal.crossing if (deal.current_trend, c) in trend_cross_map), None)
            if trigger:
                return await deal.trailing_enable(trigger)
        if self.next_handler:
            return await self.next_handler.handle(deal)
        return await deal.update_deal()

async def check_deal(deal):
    # 构建责任链：注意顺序就是优先级
    handler_chain = ProfitableMultiHandler(
        ProfitableSingleHandler(
            BoughtDealHandler(
                NewDealHandler()
            )
        )
    )
    return await handler_chain.handle(deal)

方案选择建议

规则表驱动：适合快速重构，代码简洁，新增规则只需加条目，学习成本低，推荐新手使用。
Strategy模式：适合OOP风格的项目，每个场景逻辑独立，便于单元测试和扩展新状态。
责任链模式：适合规则有明确优先级，且后续需要动态调整规则顺序或新增规则节点的场景。

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

如何解决Deal类业务逻辑中的IF嵌套地狱问题？

问题回顾 #

核心规则 #

重构方案 #

方案1：规则表驱动（最易实现，推荐新手） #

方案2：Strategy模式（面向对象风格） #

方案3：责任链模式（规则优先级明确时使用） #

方案选择建议 #