基于DFS的最大收益路径求解效率优化及最优路径回溯问题

给定价格数组，智能体在每个时间点可执行以下操作：

• 若资金充足则开仓：资金减少当前价格，该价格加入持仓列表；
• 若有持仓则平仓（平最早开仓的仓位）：资金增加当前价格，移除对应持仓，收益增加当前价与开仓价的差值；
• 持有（不操作）。

目标是找到最优开平仓时机以获取最大收益。目前用DFS实现了逻辑，但处理16个观测值耗时达1分钟，想知道优化方向；另外希望获取最优路径本身，但无法存储所有路径，想了解对应的回溯算法。

以下是Python实现代码：

import matplotlib.pyplot as plt
from collections import deque
import numpy as np
import time

def optimal_strategy(prices, root):
    s = deque([root])
    m = 0.0
    while s:
        n = s.pop()
        t = n['time'] + 1
        if t == len(prices):
            m = np.max([m, n['profit']])
            continue
        p = prices[t]
        s.append({'name': 'h' + str(t), 'time': t, 'parent': n['name'],
                  'funds': n['funds'], 'positions': n['positions'], 'profit': n['profit']})
        if p < n['funds']:
            s.append({'name': 'ol' + str(t), 'time': t, 'parent': n['name'],
                      'funds': n['funds'] - p, 'positions': n['positions'] + [p], 'profit': n['profit']})
        if len(n['positions']) > 0:
            s.append({'name': 'cl' + str(t), 'time': t, 'parent': n['name'],
                      'funds': n['funds'] + p, 'positions': n['positions'][1:], 'profit': n['profit'] + p - n['positions'][0]})

    return m

nobs = 16
np.random.seed(1)
prices = np.cumsum(np.random.normal(size=nobs))
plt.plot(prices)

t0 = time.time()
m = optimal_strategy(prices, {'name': 'r', 'time': -1, 'parent': None,
                              'funds': 4.0, 'positions': [], 'profit': 0.0})
print('Time {} Max {}'.format(time.time() - t0, m))

1. 动态规划（DP）去重剪枝 #

当前DFS会遍历所有可能路径，存在大量重复状态。可以用动态规划记录每个时间点、持仓状态、剩余资金下的最大收益，避免重复计算：

• 定义状态dp[t][pos_tuple][funds]：第t时刻，持仓队列（用元组存储，保证可哈希）为pos_tuple，剩余资金为funds时的最大收益。
• 每个状态只保留当前最优的收益值，后续遇到相同状态但收益更低的直接跳过，无需继续遍历。

2. 状态维度压缩 #

当前用列表存储持仓，导致状态空间爆炸。可做如下优化：

• 由于平仓总是平最早开仓的仓位，持仓是先进先出结构，可用元组替代列表存储持仓（元组不可变，可作为哈希键）。
• 进一步简化：若只关注收益，可记录持仓数量、累计开仓总成本，结合当前价格计算平仓收益，但需注意单个仓位的独立收益特性，此方法适用于批量平仓场景，单个平仓需保留每个开仓价的话，元组仍是更稳妥的选择。

3. 提前分支剪枝 #

在DFS遍历中，提前判断当前路径的最大潜在收益是否能超过已知全局最大值，若不可能则直接剪枝：

• 计算剩余时间内的最高价格，当前持仓每个仓位的最大可能收益为「最高价格 - 开仓价」，加上当前收益总和，若小于已知最大收益，直接终止该分支遍历。

4. 数据结构轻量化 #

• 用元组或自定义类替代字典存储状态，减少内存开销和对象创建时间。例如用元组(time, funds, positions_tuple, profit, parent)，比字典更高效。
• 避免每次创建新的持仓列表，改用元组拼接（如n['positions'] + (p,)）替代列表追加，保证状态的不可变性。

5. 迭代加深DFS优化内存 #

对于更长的价格序列，迭代加深DFS可以控制内存占用，按深度逐层遍历，结合剪枝策略减少无效路径的探索。

无需存储所有路径，只需在遍历过程中记录每个最优状态的前驱节点，最后从收益最大的终点状态回溯到初始状态即可：

实现步骤 #

1. 记录前驱状态：在DP或DFS中，不仅记录每个状态的最大收益，还要保存达到该状态的前驱状态及对应的操作（开仓/平仓/持有）。
2. 追踪最优终点：遍历到最后一个时间点时，找到收益最大的状态作为回溯起点。
3. 反向回溯路径：从终点状态开始，不断查找前驱状态，直到回到初始状态，最后将路径反转得到正向操作序列。

简化修改示例 #

def optimal_strategy_with_path(prices, root):
    s = deque([root])
    max_profit = 0.0
    best_final_state = None
    # 存储状态到最大收益、前驱、操作的映射，状态键用(time, funds, 持仓元组, profit)
    state_map = {}

    while s:
        n = s.pop()
        t = n['time'] + 1
        # 生成当前状态的唯一键
        state_key = (n['time'], round(n['funds'], 4), tuple(n['positions']), round(n['profit'], 4))
        # 若当前状态已记录且收益更低，直接跳过
        if state_key in state_map and state_map[state_key]['profit'] >= n['profit']:
            continue
        state_map[state_key] = {
            'profit': n['profit'],
            'parent': n['parent'],
            'operation': n.get('operation', 'init')
        }

        if t == len(prices):
            if n['profit'] > max_profit:
                max_profit = n['profit']
                best_final_state = state_key
            continue
        p = prices[t]
        # 持有操作
        hold_state = {
            'name': 'h' + str(t), 'time': t, 'parent': state_key,
            'funds': n['funds'], 'positions': n['positions'], 'profit': n['profit'],
            'operation': 'hold'
        }
        s.append(hold_state)
        # 开仓操作
        if p < n['funds']:
            open_state = {
                'name': 'ol' + str(t), 'time': t, 'parent': state_key,
                'funds': n['funds'] - p, 'positions': n['positions'] + [p], 'profit': n['profit'],
                'operation': 'open'
            }
            s.append(open_state)
        # 平仓操作
        if len(n['positions']) > 0:
            close_profit = n['profit'] + p - n['positions'][0]
            close_state = {
                'name': 'cl' + str(t), 'time': t, 'parent': state_key,
                'funds': n['funds'] + p, 'positions': n['positions'][1:], 'profit': close_profit,
                'operation': 'close'
            }
            s.append(close_state)

    # 回溯生成路径
    path = []
    current_key = best_final_state
    while current_key is not None:
        path.append(state_map[current_key]['operation'])
        current_key = state_map[current_key]['parent']
    path.reverse()
    return max_profit, path[1:]  # 移除初始的init操作

内容的提问来源于stack exchange，提问作者danas.zuokas