AKQuant

import akquant as aqimport akshare as akfrom akquant import Strategy
# 1. 准备数据# 使用 akshare 获取 A 股历史数据 (需安装: pip install akshare)df = ak.stock_zh_a_daily(symbol="sh600000", start_date="20250212", end_date="20260212")# 交易日历: 不传时按工作日推算, 节假日会被当作交易日trade_dates = ak.tool_trade_date_hist_sina()["trade_date"]

class MyStrategy(Strategy):    def on_bar(self, bar):        # 简单策略示例:        # 当收盘价 > 开盘价 (阳线) -> 买入        # 当收盘价 < 开盘价 (阴线) -> 卖出
        # 获取当前持仓        current_pos = self.get_position(bar.symbol)
        if current_pos == 0 and bar.close > bar.open:            self.buy(symbol=bar.symbol, quantity=100)            print(f"[{bar.timestamp_iso}] Buy 100 at {bar.close:.2f}")  # UTC ISO 8601
        elif current_pos > 0 and bar.close < bar.open:            self.close_position(symbol=bar.symbol)            print(f"[{bar.timestamp_iso}] Sell 100 at {bar.close:.2f}")  # UTC ISO 8601

# 运行回测result = aq.run_backtest(    data=df,    strategy=MyStrategy,    initial_cash=100000.0,    symbols="sh600000",    broker_profile="cn_stock_t1_low_fee",  # A 股佣金/印花税/过户费、T+1、一手 100 股    trading_calendar=trade_dates,)
# 打印回测结果print("\n=== Backtest Result ===")print(result)
# 生成最小基准对比报告benchmark_returns = (    df.set_index("date")["close"].pct_change().fillna(0.0).rename("SIMPLE_BENCH"))result.viz.report(    filename="quickstart_report.html",    show=False,    benchmark=benchmark_returns,)

from akquant import OrderStatus, Strategy
class BracketHelperStrategy(Strategy):    def __init__(self):        self.entry_order_id = ""
    def on_bar(self, bar):        if self.get_position(bar.symbol) > 0 or self.entry_order_id:            return
        self.entry_order_id = self.place_bracket(            symbol=bar.symbol,            quantity=100,            stop_trigger_price=bar.close * 0.98,            take_profit_price=bar.close * 1.04,            entry_tag="entry",            stop_tag="stop",            take_profit_tag="take",        )
    def on_order(self, order):        if order.id == self.entry_order_id and order.status in (            OrderStatus.Cancelled,            OrderStatus.Rejected,        ):            self.entry_order_id = ""

def on_event(event):    if event["event_type"] == "finished":        payload = event["payload"]        print("status:", payload.get("status"))        print("callback_error_count:", payload.get("callback_error_count"))
result = aq.run_backtest(    data=df,    strategy=MyStrategy,    symbols="sh600000",    on_event=on_event,    show_progress=False,    stream_progress_interval=10,    stream_equity_interval=10,    stream_batch_size=32,    stream_max_buffer=256,    stream_error_mode="continue",)

# 生成交互式 HTML 报告，自动在浏览器中打开result.viz.report(    show=True,    compact_currency=True,  # 金额列按 K/M/B 紧凑显示（默认 True）)
# 如果你希望金额列保留原始数值精度（不缩写），可关闭：result.viz.report(    show=False,    filename="report_raw_amount.html",    compact_currency=False,)

exposure = result.exposure_df()             # 暴露分解（净暴露/总暴露/杠杆）attr_by_symbol = result.attribution_df(by="symbol")attr_by_tag = result.attribution_df(by="tag")capacity = result.capacity_df()             # 容量代理（成交率/换手等）orders_by_strategy = result.orders_by_strategy()         # 按策略归属聚合订单executions_by_strategy = result.executions_by_strategy() # 按策略归属聚合成交