ds.resample('2H', label="right").agg({
    "open":"first",
    "close":"last",
    "high":"max",
    "low":"min",
    "volume":"sum",
    "amount":"sum"
}).dropna()

ds.close.resample("2H", label="right").last().dropna()

ds = DataSource("bar1m_CN_STOCK_A").read(["000001.SZA"], start_date='2022-05-01')
# ds.set_index("date", inplace=True)
ds

def aggregate_to_period(df, period, align_dt_minute=0):
    default_agg_rules = {
        "open": "first",
        "high": "max",
        "low": "min",
        "close": "last",
        "volume": "sum",
        "amount": "sum",
        "turnover": "sum"
    }

    if isinstance(period, int):
        period_int = period
        period = str(period_int) + 'm'

    agg_rules = {}
    for column in df.columns:
        agg_rules[column] = default_agg_rules.get(column, "last")

    dt_column = "datetime" if "datetime" in df.columns else "date"
    if "trading_day" not in df.columns:
        df['trading_day'] = df['date'].dt.date

    first_dt = df[dt_column].iloc[0]
    if align_dt_minute > 0 and first_dt.minute != 1 and first_dt.minute != 31:
        for index, dt in enumerate(df[dt_column]):
            if (dt.minute == 1 and align_dt_minute
                    == 1) or (dt.minute == 31 and align_dt_minute == 31):
                df = df[index:]
                break

    day_df_list = []
    for _, day_df in df.groupby("trading_day"):
        idx = list(range(
            0, (len(day_df) + period_int) // period_int)) * period_int
        idx.sort()
        day_df["idx"] = idx[:len(day_df)]
        day_df = day_df.groupby("idx").agg(agg_rules)
        day_df.set_index(dt_column, inplace=True)
        day_df.reset_index(inplace=True)
        day_df_list.append(day_df)
    # xperiod_df["period"] = period
    xperiod_df = pd.concat(day_df_list)

    return xperiod_df

def aggregate_df(df, period, align_dt_minute=0):
    if isinstance(period, int):
        period_int = period
        period = str(period_int) + 'T'
    if period_int <= 1:
        return df

    symbol_key = "symbol" if "symbol" in df.columns else "instrument"
    dt_column = "datetime" if "datetime" in df.columns else "date"
    if len(df[symbol_key].unique()) == 1:
        return aggregate_to_period(df,
                                   period_int,
                                   align_dt_minute=align_dt_minute)

    symbol_dfs = []
    for _, symbol_df in df.groupby(symbol_key):
        symbol_df = aggregate_to_period(
            symbol_df, period_int, align_dt_minute=align_dt_minute)
        symbol_dfs.append(symbol_df)

    ret_df = pd.concat(symbol_dfs)
    ret_df.sort_values(dt_column, inplace=True)
    ret_df.reset_index(inplace=True)
    return ret_df

aggregate_df(ds, 120)

# 本代码由可视化策略环境自动生成 2022年5月20日 18:55
# 本代码单元只能在可视化模式下编辑。您也可以拷贝代码，粘贴到新建的代码单元或者策略，然后修改。


# 交易引擎：初始化函数，只执行一次
def m1_initialize_bigquant_run(context):
    # 加载预测数据
    context.ins = context.instruments[0]
    

# 交易引擎：每个单位时间开盘前调用一次。
def m1_before_trading_start_bigquant_run(context, data):
    # 盘前处理，订阅行情等
    context.subscribe(context.ins)
#     context.subscribe_bar(context.ins, "1m")
    context.subscribe_bar(context.ins, "11m")

# 交易引擎：tick数据处理函数，每个tick执行一次
def m1_handle_tick_bigquant_run(context, tick):
    pass

# 交易引擎：bar数据处理函数，每个时间单位执行一次
def m1_handle_data_bigquant_run(context, data):
    print(data.current_dt, data.history(context.ins, ["close","high","low","open"] , 10, '11m'))

# 交易引擎：成交回报处理函数，每个成交发生时执行一次
def m1_handle_trade_bigquant_run(context, trade):
    pass

# 交易引擎：委托回报处理函数，每个委托变化时执行一次
def m1_handle_order_bigquant_run(context, order):
    pass

# 交易引擎：盘后处理函数，每日盘后执行一次
def m1_after_trading_bigquant_run(context, data):
    pass


m2 = M.instruments.v2(
    start_date='2022-03-01',
    end_date='2022-03-05',
    market='CN_STOCK_A',
    instrument_list='000001.SZA',
    max_count=0
)

m1 = M.hftrade.v2(
    instruments=m2.data,
    start_date='',
    end_date='',
    initialize=m1_initialize_bigquant_run,
    before_trading_start=m1_before_trading_start_bigquant_run,
    handle_tick=m1_handle_tick_bigquant_run,
    handle_data=m1_handle_data_bigquant_run,
    handle_trade=m1_handle_trade_bigquant_run,
    handle_order=m1_handle_order_bigquant_run,
    after_trading=m1_after_trading_bigquant_run,
    capital_base=1000000,
    frequency='minute',
    price_type='真实价格',
    product_type='股票',
    before_start_days='10',
    order_price_field_buy='open',
    order_price_field_sell='close',
    benchmark='000300.HIX',
    plot_charts=True,
    disable_cache=False,
    replay_bdb=False,
    show_debug_info=False,
    backtest_only=False
)