import minibt
print(minibt.__version__)

from minibt import *  # 导入minibt核心功能（策略、指标、回测引擎等）

class MA(Strategy):  # 定义策略类，必须继承Strategy基类
    params = dict(
       length1=10,  # 短期均线周期（可通过参数优化调整）
       length2=20   # 长期均线周期（可通过参数优化调整）
   )
   def __init__(self):
       # 1. 加载数据：使用minibt内置测试数据（无需手动准备）
       self.data = self.get_kline(LocalDatas.test) 
      
       # 2. 计算均线指标：基于收盘价计算SMA
       self.ma1 = self.data.close.sma(self.params.length1)  # 短期均线
       self.ma2 = self.data.close.sma(self.params.length2)  # 长期均线
      

       # 3. 生成交易信号：金叉（多头）、死叉（空头）
       self.long_signal = self.ma1.cross_up(self.ma2)  # 短期均线上穿长期均线
       self.short_signal = self.ma2.cross_down(self.ma1)  # 长期均线下穿短期均线
       
   def next(self):
       # next 方法：逐根K线处理核心逻辑，策略的"大脑"
       # 情况1：无持仓时，根据信号开仓
       if not self.data.position:  # position=0 表示空仓
           if self.long_signal.new:  # 金叉信号最新值
               self.data.buy()       # 开多仓
           elif self.short_signal.new:  # 死叉信号最新值
               self.data.sell()      # 开空仓
      
       # 情况2：持有多头时，死叉信号平仓
       elif self.data.position > 0 and self.short_signal.new: 
           self.data.sell()  # 平多仓
      
       # 情况3：持有空头时，金叉信号平仓
       elif self.data.position < 0 and self.long_signal.new: 
           self.data.buy()   # 平空仓

if __name__ == "__main__":
   # 初始化回测引擎：auto=True 自动处理数据加载、回测、可视化
   bt = Bt(auto=True) 
   bt.run()  # 启动回测，完成后自动弹出可视化界面

from minibt import *

if __name__ == "__main__":
    LocalDatas.test.kline.run(
        ["ma1", Multiply(PandasTa.sma, dict(length=20))],
        ["ma2", Multiply(PandasTa.sma, dict(length=30))],
        ["ma3", Multiply(PandasTa.sma, dict(length=60))],
        ["ma4", Multiply(PandasTa.sma, dict(length=120))],
        ["ebsw", Multiply(PandasTa.ebsw)])

if __name__ == "__main__":
    bt = Bt(auto=True)
    # 设置优化目标和参数范围
    bt.optstrategy(
        ["profit", "win_rate"],  # 优化目标：收益和胜率
        (1, 1),                  # 权重配置
        length1=(5, 15, 1),      # length1参数范围：5到15，步长为1
        length2=(20, 30, 1),     # length2参数范围：20到30，步长为1
        opconfig=OptunaConfig(n_trials=100)  # Optuna配置：100次试验
    )
    bt.run()

# 获取数据值数组
values = indicator.values
# 获取数据形状（行数, 列数）
shape = indicator.shape
# 获取行数（时间步数量）
length = indicator.V  # 或 indicator.length
# 获取列数（特征维度）
columns = indicator.H

# 支持多种索引方式
indicator[0]          # 第一行
indicator['close']    # close列
indicator[10:20]      # 切片
indicator[[1, 3, 5]]  # 索引列表

# 基于位置的索引
indicator.iloc[0]        # 第一行
indicator.iloc[0:5]      # 前5行
indicator.iloc[0, 1]     # 第一行第二列

# 基于标签的索引
indicator.loc['close']   # close列
indicator.loc[0:5]       # 前5行

# 快速标量访问
indicator.at[0, 'close'] # 第一行的close值
indicator.iat[0, 1]      # 第一行第二列的值

# 基本算术运算
result = ma5 + ma10
result = close - open
result = volume * 2
result = close / open
# 原地运算
ma5 += 10
volume *= 1.5

# 比较运算
signal = close > ma20
signal = volume >= volume.mean()
signal = rsi == 50

# 逻辑运算
combined_signal = (close > ma20) & (volume > volume.mean())
either_signal = (rsi > 70) | (rsi < 30)

# 访问历史数据
current_value = indicator.new        # 当前值
previous_value = indicator.prev      # 前一个值
value_10_periods_ago = indicator.history(10)  # 10周期前的值
last_10_values = indicator.history(0, 10)     # 最近10个值

# 生成填充数据
ones = indicator.ones        # 全1序列
zeros = indicator.zeros      # 全0序列
filled = indicator.full(100) # 填充指定值的序列

# 多条件数据处理
signal = indicator.ifs(
    condition1, value1,    # 条件1满足时设为value1
    condition2, value2,    # 条件2满足时设为value2
    other=default_value    # 默认值
)

# 示例：生成交易信号
long_signal = price.ifs(
    price > ma20, 1,       # 价格上穿MA20时设为1
    price < ma10, 0,       # 价格下穿MA10时设为0
    other=0                # 默认值0（无信号）
)

# 滚动窗口计算
rolling_mean = indicator.rolling(window=20).mean()
rolling_std = indicator.rolling(window=20).std()
rolling_max = indicator.rolling(window=20).max()

# 自定义滚动函数
def custom_roll_func(window_data):
    return window_data.max() - window_data.min()

rolling_range = indicator.rolling_apply(custom_roll_func, window=20)

# 指标上采样
minute_data = indicator.upsample()  # 默认参数
minute_data = indicator()

# 多指标并行计算
result1, result2, result3 = indicator.multi_apply(
    [func1, params1],  # 第一个指标
    [func2, params2],  # 第二个指标
    [func3, params3]   # 第三个指标
)

# 使用PandasTA指标
pandas_ta_result = indicator.pta.sma(length=20)

# 使用TA-Lib指标
talib_result = indicator.talib.RSI(timeperiod=14)

# 使用Tulip指标
tulip_result = indicator.tulip.wma(period=10)

# 使用内置指标
btind_result = indicator.btind.pmax(period=14, multiplier=3)

# 使用FinTa指标
finta_result = indicator.finta.SMA(period=20)

# 使用天勤指标
tqfunc_result = indicator.tqfunc.barlast()

# 使用天勤指标
tqta_result = indicator.tqta.RSI()

# 使用信号特征
signal_features = indicator.sf.Binarizer()

# 使用配对交易指标
pair_result = indicator.pairtrading.bollinger_bands()

# 使用因子分析
factors: pd.DataFrame
factor_result = indicator.factors.single_asset_multi_factor_strategy(*factors)

# 使用TradingView指标
gc = btdata.tradingview.G_Channels()

# 设置指标名称
indicator.sname = "my_indicator"           #策略里面命名的变量名称
indicator.ind_name = "Custom Indicator"    #指标名称

# 设置绘图开关
indicator.isplot = True  # 全部显示
indicator.isplot = {"line1": True, "line2": False}  # 按线条名设置

# 设置主图叠加
indicator.overlap = True  # 全部叠加
indicator.overlap = {"line1": True, "line2": False}  # 按线条名设置

# 设置绘图高度
indicator.height = 300  # 指标图表显示高度

# 添加水平线
indicator.span_style = 50.0  # 单个水平线
indicator.span_style = SpanStyle(  # 完整配置
    location=50.0,
    color="red",
    dash="dashed",
    width=2.0
)

# 检查数据类型
is_indicator = indicator.isindicator  # 是否为指标
is_custom = indicator.iscustom        # 是否为自定义数据
is_mdim = indicator.isMDim            # 是否为多维数据
is_main = indicator.ismain            # 是否在主周期显示
is_resample = indicator.isresample    # 是否为转换数据
is_replay = indicator.isreplay        # 是否为回放数据
is_candles = indicator.iscandles      # 是否为蜡烛图
is_signal = indicator.issignal        # 是否包含交易信号

# 获取关联的策略和数据
strategy = indicator.strategy_instance  # 关联的策略实例
bt_data = indicator.kline               # 关联的K线数据

# 获取策略ID和索引
strategy_id = indicator.strategy_id    # 策略ID
plot_id = indicator.plot_id            # 绘图ID
data_id = indicator.data_id            # 数据ID
resample_id = indicator.resample_id    # 转换ID
replay_id = indicator.replay_id        # 回放ID

# 获取回测状态
current_index = indicator.btindex      # 当前回测索引
is_live = indicator.islivetrading      # 是否为实盘交易

from minibt import *


class MultiTimeframeStrategy(Strategy):
    def __init__(self):
        # 主数据（1分钟）
        self.data_1m = self.get_kline(LocalDatas.v2509_60)
        # 转换到5分钟数据
        self.data_5m = self.data_1m.resample(300)
        # 在不同周期上计算指标
        self.ma_1m = self.data_1m.close.sma(20)
        # 数据上采样
        self.ma_5m = self.data_5m.close.sma(20)()
        self.ma_5m.overlap = True
        # 交易信号
        self.long_signal = self.ma_1m.cross_up(self.ma_5m)
        self.short_signal = self.ma_1m.cross_down(self.ma_5m)

    def next(self):
        if not self.data_1m.position:
            # 多周期信号组合
            if self.long_signal.new:
                self.buy()
            elif self.short_signal.new:
                self.sell()
        elif self.data_1m.position > 0 and self.short_signal.new:
            self.sell()
        elif self.data_1m.position < 0 and self.long_signal.new:
            self.buy()


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

from minibt import *


class G_Channels(BtIndicator):
    """G-Channels指标 - 高效计算上下极值通道"""
    params = dict(length=144., cycle=1)
    overlap = True  # 主图叠加显示

    def next(self):
        length = self.params.length
        cycle = min(max(int(self.params.cycle), 1), 3)
        size = self.close.size
        a = self.full()  # 上通道
        b = self.full()  # 下通道
        close = self.close.values
        pre_a = 0.
        pre_b = 0.
        # 计算通道
        for i in range(size):
            if i and i % cycle == 0:
                a[i] = max(close[i], pre_a) - (pre_a - pre_b) / length
                b[i] = min(close[i], pre_b) + (pre_a - pre_b) / length
                pre_a = a[i]
                pre_b = b[i]
        # 周期大于1时进行插值
        if cycle > 1:
            a = a.interpolate()
            b = b.interpolate()
        # 计算中轨
        avg = (a + b) / 2.
        return a, b, avg  # 返回上轨、下轨、中轨


class MyStrategy(Strategy):
    def __init__(self):
        self.data = self.get_kline(LocalDatas.v2509_60)
        self.data.height = 600
        self.G_Channels = G_Channels(self.data)
        self.G_Channels.a.line_style = LineStyle(
            LineDash.solid, 2, Colors.blue),     # 上轨样式
        self.G_Channels.b.line_style = LineStyle(
            LineDash.solid, 2, Colors.red),      # 下轨样式
        self.G_Channels.avg.line_style = LineStyle(
            LineDash.dotted, 3, Colors.gray)     # 中轨样式


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

from __future__ import annotations
from minibt import *
from tqsdk import TqApi, TqAuth, TqKq


def get_contract(exchange_id: str, contracts: list[str], has_night=None):
    global api
    target_prefixes: list[str] = [
        f"{exchange_id}.{cont}" for cont in contracts]
    all_contracts: list[str] = api.query_cont_quotes(
        exchange_id, has_night=has_night)

    return [
        contract for contract in all_contracts
        for prefix in target_prefixes
        if contract.startswith(prefix) and contract[len(prefix):].isdigit()
    ]


class MyStrategy(Strategy):
    params = dict(symbol="DCE.v2601")

    def __init__(self) -> None:
        self.data = self.get_kline(self.params.symbol, 60, 1000)
        self.data.watermark = True
        self.toolbox = False
        self.gc = self.data.tradingview.G_Channels()
        self.ha = self.data.ha()
        self.pmax = self.data.close.btind.pmax3().pmax
        self.macd = self.data.close.macd()
        self.macd.macdh.line_color = Colors.limegreen
        self.macd.macdh.price_label = True
        self.macd.macdh.price_line = True


if __name__ == "__main__":
    api = TqApi(TqKq(), auth=TqAuth(
        "天勤账户", "天勤密码"))
    bt = Bt(auto=False, live=True)
    bt.addTqapi(api=api)

    contracts_dict = {
        "SHFE": ["cu", "ag"],
        # "DCE": ["v", "pp",  "m", "l", "eg", "eb", "b"],
        # "CZCE": ["TA", "SR", "SM", "RM", "PX", "PF", "MA"]
    }
    name = "策略"
    i = 0
    for k, v in contracts_dict.items():
        contracts = get_contract(k, v)
        for contract in contracts:
            i += 1
            bt.addstrategy(
                MyStrategy.copy(name=f"{name}{i}", params=dict(symbol=contract)))

    bt.run(period_milliseconds=1000)