工作流自动化

def signal_message(data):
    close = data.close.iloc[:-1]  
    fast = vbt.MA.run(close, window=20)
    slow = vbt.MA.run(close, window=50)
    bar = f"{close.index[-1]:%Y-%m-%d %H:%M} BTCUSDT 20h MA crossed"
    if fast.ma_crossed_above(slow).iloc[-1]:
        return f"{bar} above 50h MA at {close.iloc[-1]:,.0f}"
    if fast.ma_crossed_below(slow).iloc[-1]:
        return f"{bar} below 50h MA at {close.iloc[-1]:,.0f}"

class AlertUpdater(vbt.DataUpdater):
    def __init__(self, data, notify, **kwargs):
        vbt.DataUpdater.__init__(self, data, notify=notify, **kwargs)
        self._notify = notify

    def update(self, **kwargs):
        vbt.DataUpdater.update(self, **kwargs)  
        message = signal_message(self.data)
        if message is not None:
            self._notify(message)

data = vbt.BinanceData.pull(
    "BTCUSDT", timeframe="1h", start="2024-11-24", end="2024-12-01 00:01"
)
updater = AlertUpdater(data, notify=print)
for hour in pd.date_range("2024-12-01 01:00", periods=7 * 24, freq="h", tz="UTC"):
    updater.update(end=hour + pd.Timedelta(minutes=1))  

bot = vbt.TelegramBot(token="YOUR_BOT_TOKEN")  
bot.start(in_background=True)

updater = AlertUpdater(data, notify=bot.send_message_to_all)  
updater.update_every("hour", "01:00")  

import calendar

data = vbt.YFData.pull("BTC-USD")

@vbt.iterated(over_arg="year", merge_func="column_stack", engine="pathos")  
@vbt.iterated(over_arg="month", merge_func="concat")  
def get_year_month_sharpe(data, year, month):  
    mask = (data.index.year == year) & (data.index.month == month)
    if not mask.any():
        return np.nan
    year_returns = data.loc[mask].returns
    return year_returns.vbt.returns.sharpe_ratio()

years = data.index.year.unique().sort_values().rename("year")
months = data.index.month.unique().sort_values().rename("month")
sharpe_matrix = get_year_month_sharpe(
    data,
    years,
    {calendar.month_abbr[month]: month for month in months},  
)
sharpe_matrix.transpose().vbt.heatmap(
    trace_kwargs=dict(colorscale="RdBu", zmid=0),
    yaxis=dict(autorange="reversed")
).show()

data = vbt.YFData.pull("BTC-USD")

task1 = vbt.Task(  
    vbt.PF.from_random_signals,
    data,
    n=100, seed=42,
    sl_stop=vbt.Param(np.arange(1, 51) / 100)
)
task2 = vbt.Task(
    vbt.PF.from_random_signals,
    data,
    n=100, seed=42,
    tsl_stop=vbt.Param(np.arange(1, 51) / 100)
)
task3 = vbt.Task(
    vbt.PF.from_random_signals,
    data,
    n=100, seed=42,
    tp_stop=vbt.Param(np.arange(1, 51) / 100)
)
pf1, pf2, pf3 = vbt.execute([task1, task2, task3], engine="pathos")  

fig = pf1.trades.expectancy.rename("SL").vbt.plot()
pf2.trades.expectancy.rename("TSL").vbt.plot(fig=fig)
pf3.trades.expectancy.rename("TP").vbt.plot(fig=fig)
fig.show()

symbols = ["BTC-USD", "ETH-USD"]
fast_windows = range(5, 105, 5)
slow_windows = range(5, 105, 5)
sharpe_ratios = dict()

with vbt.ProgressBar(total=len(symbols), bar_id="pbar1") as pbar1:  
    for symbol in symbols:
        pbar1.set_description(dict(symbol=symbol), refresh=True)
        data = vbt.YFData.pull(symbol)

        with vbt.ProgressBar(total=len(fast_windows), bar_id="pbar2") as pbar2:  
            for fast_window in fast_windows:
                pbar2.set_description(dict(fast_window=fast_window), refresh=True)

                with vbt.ProgressBar(total=len(slow_windows), bar_id="pbar3") as pbar3:  
                    for slow_window in slow_windows:
                        if fast_window < slow_window:
                            pbar3.set_description(dict(slow_window=slow_window), refresh=True)
                            fast_sma = data.run("talib_func:sma", fast_window)
                            slow_sma = data.run("talib_func:sma", slow_window)
                            entries = fast_sma.vbt.crossed_above(slow_sma)
                            exits = fast_sma.vbt.crossed_below(slow_sma)
                            pf = vbt.PF.from_signals(data, entries, exits)
                            sharpe_ratios[(symbol, fast_window, slow_window)] = pf.sharpe_ratio
                        pbar3.update()

                pbar2.update()

        pbar1.update()

sharpe_ratios = pd.Series(sharpe_ratios)
sharpe_ratios.index.names = ["symbol", "fast_window", "slow_window"]
sharpe_ratios