带你详细了解PTrade策略框架（第五篇）after_trading_end函数详解

我们已经依次学习了策略的初始化(initialize)、盘前准备(before_trading_start)和核心交易(handle_data)重要API函数

本文将介绍PTrade策略生命周期的最后一个核心函数——after_trading_end()。为当天的交易画上句号，为交易做好准备

1. after_trading_end(context) 函数的核心定位

after_trading_end()函数会在每个交易日的收盘后被平台自动调用,它的核心任务是执行每日的复盘、数据统计和持久化存储工作

在这个函数中，当天的所有交易均已完成，市场已经关闭，因此它是一个进行静态分析和数据整理的理想时机

参数 context：作为当天最后一个被调用的函数，context对象包含了策略当日最终的账户状态，如收盘后的总资产、现金和最终持仓情况

2. after_trading_end() 的核心任务

after_trading_end()的职责清晰地聚焦于“盘后”二字，主要包括以下几个方面：

①每日业绩复盘与记录

最核心的用途就是记录和分析当天的策略表现。我们可以利用这个函数，将关心的各项指标输出到日志中

以便我们后续进行更详细的分析。做好下一次的应对策略

关键信息：当日最终总资产、持仓市值、可用现金、持仓股票列表及其成本、当日盈亏等

②一些策略状态的持久化

有些策略的逻辑需要依赖前一天的计算结果。例如，一个动态调整的止损线，或者一个需要每日更新的模型参数。after_trading_end()提供了一个将这些需要“隔夜”传递的变量保存下来的机会。常见的持久化方式包括写入CSV文件、JSON文件或数据库(后面我们会详细来学习)

③生成自定义交易报告

对于希望每日监控策略状态的用户，可以在此函数中编写逻辑，将当日的交易总结（如成交记录、盈亏分析等）格式化，并发送到指定的邮箱或通过其他方式进行通知（具体实现需借助第三方库）。

3. 完整的 after_trading_end() 代码示例

def initialize(context):
    set_benchmark("000300.SS")

    try:
        set_commission(commission_ratio=0.0003, min_commission=5.0, type="STOCK")
        set_slippage(slippage=0.0024)
    except Exception as e:
        log.info(f"提示：当前环境可能不支持 set_commission/set_slippage：{e}")

    g.security_list = ["600519.SS", "000001.SZ"]
    set_universe(g.security_list)

    g.short_ma_period = 10
    g.long_ma_period = 30

    g.traded_today = False
    g.today_options = list(g.security_list)
    g._last_debug_day = None

    log.info("策略初始化完成。")


def before_trading_start(context, data):
    g.traded_today = False

    dt = getattr(context, "current_dt", None)
    date_str = dt.strftime("%Y-%m-%d") if dt else "unknown_date"
    log.info(f"进入交易日 {date_str}，重置每日状态。")

    sec_list = getattr(g, "security_list", None) or ["600519.SS", "000001.SZ"]
    g.security_list = sec_list

    g.today_options = []

    try:
        hist = get_history(1, "1d", "close", security_list=sec_list, fq=None, include=False)
    except TypeError:
        hist = get_history(1, frequency="1d", field="close", security_list=sec_list, fq=None, include=False)

    if hist is None:
        g.today_options = list(sec_list)
    else:
        close_map = None
        try:
            import pandas as pd
            if isinstance(hist, pd.DataFrame):
                cols = set(hist.columns)
                if ("code" in cols) and ("close" in cols):
                    m = {}
                    for s in sec_list:
                        x = hist.loc[hist["code"] == s, "close"]
                        if len(x) > 0:
                            m[s] = float(x.iloc[-1])
                    close_map = m
                elif all(s in cols for s in sec_list):
                    close_map = {s: float(hist[s].iloc[-1]) for s in sec_list}
            elif isinstance(hist, pd.Series):
                close_map = {sec_list[0]: float(hist.iloc[-1])}
        except Exception:
            close_map = None

        if close_map is None:
            g.today_options = list(sec_list)
        else:
            for s in sec_list:
                if s in close_map and close_map[s] > 20:
                    g.today_options.append(s)
            if not g.today_options:
                g.today_options = list(sec_list)

    log.info(f"今日备选股票池: {g.today_options}")
    log.info(f"盘前可用资金: {context.portfolio.cash:.2f}")


def _extract_close_series(hist, security):
    try:
        import pandas as pd
        if isinstance(hist, pd.Series):
            return hist.dropna()
        if isinstance(hist, pd.DataFrame):
            cols = set(hist.columns)
            if ("code" in cols) and ("close" in cols):
                return hist.loc[hist["code"] == security, "close"].dropna()
            if security in cols:
                return hist[security].dropna()
            if "close" in cols:
                return hist["close"].dropna()
    except Exception:
        pass

    try:
        return hist["close"][security].dropna()
    except Exception:
        return None


def handle_data(context, data):
    sec_list = getattr(g, "security_list", None) or ["600519.SS", "000001.SZ"]
    g.security_list = sec_list

    short_n = int(getattr(g, "short_ma_period", 10))
    long_n = int(getattr(g, "long_ma_period", 30))
    if not (0 < short_n < long_n):
        log.info("均线参数不合法：要求 0 < short < long。")
        return

    candidates = getattr(g, "today_options", None) or sec_list
    if not candidates:
        return
    security = candidates[0]

    try:
        hist = get_history(long_n + 1, "1d", "close", security_list=security, fq=None, include=False)
    except TypeError:
        hist = get_history(long_n + 1, frequency="1d", field="close", security_list=security, fq=None, include=False)

    if hist is None:
        return

    close = _extract_close_series(hist, security)
    if close is None or len(close) < long_n + 1:
        log.info(f"{security} 历史数据不足或解析失败，跳过。")
        return

    s_ma_t = close.iloc[-short_n:].mean()
    l_ma_t = close.iloc[-long_n:].mean()

    close_prev = close.iloc[:-1]
    s_ma_prev = close_prev.iloc[-short_n:].mean()
    l_ma_prev = close_prev.iloc[-long_n:].mean()

    golden_cross = (s_ma_prev  l_ma_t)
    death_cross = (s_ma_prev >= l_ma_prev) and (s_ma_t < l_ma_t)

    pos = get_position(security)
    pos_amount = getattr(pos, "amount", 0) or 0

    dt = getattr(context, "current_dt", None)
    d = dt.date() if dt else None
    if getattr(g, "_last_debug_day", None) != d:
        g._last_debug_day = d
        log.info(
            f"{security} | s_prev={s_ma_prev:.3f}, l_prev={l_ma_prev:.3f}, "
            f"s={s_ma_t:.3f}, l={l_ma_t:.3f} | "
            f"golden={golden_cross}, death={death_cross} | pos_amount={pos_amount}"
        )

    try:
        current_price = data[security].price
    except Exception:
        try:
            current_price = data[security]["close"]
        except Exception:
            current_price = None

    target_amount = 100

    if golden_cross and pos_amount == 0:
        delta = target_amount - pos_amount
        if delta != 0:
            if current_price is not None:
                oid = order(security, delta, limit_price=current_price)
            else:
                oid = order(security, delta)
            log.info(f"金叉下单：{security} delta={delta} price={current_price} order_id={oid}")
            g.traded_today = True

    elif death_cross and pos_amount > 0:
        delta = -pos_amount
        if current_price is not None:
            oid = order(security, delta, limit_price=current_price)
        else:
            oid = order(security, delta)
        log.info(f"死叉下单：{security} delta={delta} price={current_price} order_id={oid}")
        g.traded_today = True


def after_trading_end(context, data):
    log.info("======================================================================")

    dt = getattr(context, "current_dt", None)
    date_str = dt.strftime("%Y-%m-%d") if dt else "unknown_date"
    log.info(f"交易日 {date_str} 收盘，开始进行每日复盘...")

    total_value = context.portfolio.total_value
    cash = context.portfolio.cash
    log.info(f"收盘总资产: {total_value:.2f}")
    log.info(f"收盘可用现金: {cash:.2f}")

    try:
        all_positions = get_all_positions()
    except Exception:
        all_positions = None

    if not all_positions:
        log.info("当日无持仓，空仓过夜。")
    else:
        log.info("当日最终持仓详情:")
        for security, position in all_positions.items():
            total_amt = getattr(position, "total_amount", None)
            if total_amt is None:
                total_amt = getattr(position, "amount", 0) or 0

            avg_cost = getattr(position, "avg_cost", None)
            if avg_cost is None:
                avg_cost = getattr(position, "cost_basis", 0.0) or 0.0

            value = getattr(position, "value", None)
            if value is None:
                value = getattr(position, "market_value", 0.0) or 0.0

            log.info(
                f"  - 股票: {security}, "
                f"持有数量: {int(total_amt)}, "
                f"平均成本: {float(avg_cost):.2f}, "
                f"当前市值: {float(value):.2f}"
            )

    log.info("每日复盘结束。")
    log.info("======================================================================\n")

日志如下：

注意事项

• 禁止交易操作：如果市场已经休市，在after_trading_end()中严禁调用任何下单函数（如order），否则会导致策略错误
• 数据局限性：与before_trading_start类似，这里无法获取分钟线等盘中数据，只能获取到当日最终的日线数据
• 性能考量：盘后阶段同样需要执行结算等任务，应避免在此函数中进行过于消耗资源的计算，保证策略能够快速完成每日的收尾工作

至此，我们已经完整地学习了PTrade策略框架的四大核心函数，它们共同构成了策略的基本框架从initialize的一次性设置，到before_trading_start的每日准备，再到handle_data的盘中决策，最后由after_trading_end完成每日复盘，形成了一个完整而严谨的闭环

从下一系列文章开始，我们将深入PTrade的各类API函数及具体用法，例如如何获取更丰富的历史数据、如何使用不同类型的下单指令等，敬请期待！！

PTrade免费申请

PTradeQMT免费领取学习案例

PTradeQMT落地辅助

需要的朋友欢迎联系 ~~~