leopard-analysis/notebook/backtest.py

import marimo

__generated_with = "0.19.6"
app = marimo.App(width="full", auto_download=["ipynb"], sql_output="pandas")


@app.cell
def _():
    import marimo as mo
    return (mo,)


@app.cell
def _():
    import urllib
    import sqlalchemy

    host = "81.71.3.24"
    port = 6785
    username = "leopard"
    password = urllib.parse.quote_plus("9NEzFzovnddf@PyEP?e*AYAWnCyd7UhYwQK$pJf>7?ccFiN^x4$eKEZ5~E<7<+~X")
    database = "leopard_dev"
    engine = sqlalchemy.create_engine(f"postgresql://{username}:{password}@{host}:{port}/{database}")
    return (engine,)


@app.cell
def _(engine, mo):
    dailies_df = mo.sql(
        f"""
        select trade_date,
               open * factor         as Open,
               close * factor        as Close,
               high * factor         as High,
               low * factor          as Low,
               volume                as Volume,
               coalesce(factor, 1.0) as factor
        from leopard_daily daily
                 left join leopard_stock stock on stock.id = daily.stock_id
        where stock.code = '000001.SZ'
          and daily.trade_date between '2024-01-01 00:00:00' and '2025-12-31 23:59:59'
        order by daily.trade_date
        """,
        engine=engine
    )
    return (dailies_df,)


@app.cell
def _(dailies_df):
    import pandas as pd

    dailies_df.rename(
        columns={'open': 'Open', 'close': 'Close', 'high': 'High', 'low': 'Low', 'volume': 'Volume'}, inplace=True
    )
    dailies_df['trade_date'] = pd.to_datetime(dailies_df['trade_date'], format='%Y-%m-%d')
    dailies_df.set_index('trade_date', inplace=True)
    dailies_df
    return


@app.cell
def _(dailies_df):
    import talib

    dailies_df['sma10'] = talib.SMA(dailies_df['Close'], timeperiod=10)
    dailies_df['sma30'] = talib.SMA(dailies_df['Close'], timeperiod=30)
    dailies_df['sma60'] = talib.SMA(dailies_df['Close'], timeperiod=60)
    dailies_df['sma120'] = talib.SMA(dailies_df['Close'], timeperiod=120)
    dailies_df
    return


@app.cell
def _(dailies_df):
    # 指标计算完成后截取后面指标的完整的部份使用
    target_dailies_df = dailies_df.loc['2025-01-01':'2025-12-31']
    target_dailies_df
    return (target_dailies_df,)


@app.cell
def _():
    from backtesting import Strategy
    from backtesting.lib import crossover

    class SmaCross(Strategy):
        def init(self):
            self.sma10 = self.I(lambda x: x, self.data.sma10)
            self.sma30 = self.I(lambda x: x, self.data.sma30)
            self.sma60 = self.I(lambda x: x, self.data.sma60)
            # self.sma120 = self.I(lambda x: x, self.data.sma120)

        def next(self):
            if self.sma60 > 0 and crossover(self.data.sma10, self.data.sma30):
                self.buy()
            elif self.position.size > 0 and crossover(self.data.sma30, self.data.sma10):
                self.position.close()
    return (SmaCross,)


@app.function
def stats_print(stats):
    indicator_name_mapping = {
        # 'Start': '回测开始时间',
        # 'End': '回测结束时间',
        # 'Duration': '回测持续时长',
        # 'Exposure Time [%]': '持仓时间占比（%）',
        'Equity Final [$]': '最终收益',
        'Equity Peak [$]': '峰值收益',
        'Return [%]': '总收益率（%）',
        'Buy & Hold Return [%]': '买入并持有收益率（%）',
        'Return (Ann.) [%]': '年化收益率（%）',
        'Volatility (Ann.) [%]': '年化波动率（%）',
        # 'CAGR [%]': '复合年均增长率（%）',
        # 'Sharpe Ratio': '夏普比率',
        'Sortino Ratio': '索提诺比率',
        'Calmar Ratio': '卡尔玛比率',
        # 'Alpha [%]': '阿尔法系数（%）',
        # 'Beta': '贝塔系数',
        'Max. Drawdown [%]': '最大回撤（%）',
        'Avg. Drawdown [%]': '平均回撤（%）',
        'Max. Drawdown Duration': '最大回撤持续时长',
        'Avg. Drawdown Duration': '平均回撤持续时长',
        '# Trades': '总交易次数',
        'Win Rate [%]': '胜率（%）',
        # 'Best Trade [%]': '最佳单笔交易收益率（%）',
        # 'Worst Trade [%]': '最差单笔交易收益率（%）',
        # 'Avg. Trade [%]': '平均单笔交易收益率（%）',
        # 'Max. Trade Duration': '单笔交易最长持有时长',
        # 'Avg. Trade Duration': '单笔交易平均持有时长',
        # 'Profit Factor': '盈利因子',
        # 'Expectancy [%]': '期望收益（%）',
        'SQN': '系统质量数',
        # 'Kelly Criterion': '凯利准则',
    }
    for k, v in stats.items():
        if k in indicator_name_mapping:
            cn_name = indicator_name_mapping.get(k, k)
            if isinstance(v, (int, float)):
                if "%" in cn_name or k in ['Sharpe Ratio', 'Sortino Ratio', 'Calmar Ratio', 'Profit Factor']:
                    formatted_value = f"{v:.2f}"
                elif "$" in cn_name:
                    formatted_value = f"{v:.2f}"
                elif "次数" in cn_name:
                    formatted_value = f"{v:.0f}"
                else:
                    formatted_value = f"{v:.4f}"
            else:
                formatted_value = str(v)
            print(f'{cn_name}: {formatted_value}')


@app.cell
def _(SmaCross, target_dailies_df):
    from backtesting import Backtest

    import backtesting._plotting as plotting
    from bokeh.colors.named import tomato, lime

    plotting.BULL_COLOR = tomato
    plotting.BEAR_COLOR = lime

    bt = Backtest(target_dailies_df, SmaCross, cash=100000, commission=.002, finalize_trades=True)
    stats = bt.run()
    stats_print(stats)
    return bt, stats


@app.cell
def _(stats):
    stats._trades
    # stats._equity_curve
    return


@app.cell
def _(bt):
    bt.plot()
    return


if __name__ == "__main__":
    app.run()