Привет! Напиши мне DAG для считывания данных BTC-USDT с биржи OKX, его обработки и сохранение в InfluxDB

1) Считывание данных из OKX с помощью библиотеки ccxt за последний час. Т.е. надо считать одну точку данных - это свеча + Volume с 1 часовым таймфреймом

    # Биржа из которой будут браться данные с помощью CCXT
    EXCHANGE = ccxt.okx()
    # Инструмент в формате символа для обработки
    SYMBOL = "BTC/USDT"
    # Таймфрейм свеч
    TIMEFRAME = "1h"


2) Подгрузить 299 последних точек из InfluxDB

check_query = f'''
from(bucket:"{bucket_name}")
  |> range(start: -inf)
  |> filter(fn: (r) => r["_measurement"] == "{measurement_name}")
  |> last()
  |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
'''

result = query_api.query_data_frame(check_query)
start_query_time = result["_time"].copy() - pd.Timedelta(hours=299-1)
start_query_time = start_query_time[0]
start_query_time = start_query_time.isoformat()

last_data_query = f'''
from(bucket:"{bucket_name}")
  |> range(start: {start_query_time})
  |> filter(fn: (r) => r["_measurement"] == "{measurement_name}")
  |> filter(fn: (r) => r["_field"] == "Close" or r["_field"] == "Open" or r["_field"] == "Low" or r["_field"] == "High" or r["_field"] == "Volume")
  |> limit(n: {LAST_CLOSE_POINT_COUNT})
  |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
'''

last_data_df = query_api.query_data_frame(last_data_query)
last_data_df = last_data_df[["_time", "Open", "High", "Low", "Close", "Volume"]]
last_data_df = last_data_df.rename(columns={"_time": "Date"})

df_for_calculations = pd.concat([last_data_df, df_test.iloc[0].to_frame().T], axis=0)
df_for_calculations = df_for_calculations.reset_index(drop=True)

3.1) Добавление новых фичей на основе close данных

with pd.option_context('mode.chained_assignment', None):
        ohlcv["Close_kalman_filter"] = apply_kalman_filtering(ohlcv["Close"])
        ohlcv["Close_savitzky_golay_filter"] = apply_savitzky_golay_filtering(ohlcv["Close"])
        ohlcv["Close_wavelet_filter"] = apply_wavelet_filtering(ohlcv["Close"])
        ohlcv["Close_gaussian_smoothing"] = apply_gaussian_smoothing(ohlcv["Close"])
        ohlcv["Close_sma_smoothing"] = apply_sma_smoothing(ohlcv["Close"])
        ohlcv["Close_exponential_smoothing"] = apply_exponential_smoothing_smoothing(ohlcv["Close"])

        return ohlcv

3.2) Создание индикаторов тех анализа на основе OHLCV

with pd.option_context('mode.chained_assignment', None):
        rsi_signals = rsi_signal(ohlcv)
        ohlcv["RSI_buy_signal"] = rsi_signals[0]
        ohlcv["RSI_sell_signal"] = rsi_signals[1]

        bbands_signals = bbands_signal(ohlcv)
        ohlcv["BBANDS_buy_signal"] = bbands_signals[0]
        ohlcv["BBANDS_sell_signal"] = bbands_signals[1]

        macd_signals = macd_signal(ohlcv)
        ohlcv["MACD_buy_signal"] = macd_signals[0]
        ohlcv["MACD_sell_signal"] = macd_signals[1]

        ppo_signals = ppo_signal(ohlcv)
        ohlcv["PPO_buy_signal"] = ppo_signals[0]
        ohlcv["PPO_sell_signal"] = ppo_signals[1]

        adx_signals = adx_signal(ohlcv)
        ohlcv["ADX_buy_signal"] = adx_signals[0]
        ohlcv["ADX_sell_signal"] = adx_signals[1]

4) Сохранение в InfluxDB

global INFLUXDB_URL, INFLUXDB_TOKEN, INFLUXDB_ORG, INFLUXDB_BUCKET_NAME, INFLUXDB_MEASUREMENT_NAME

    client = InfluxDBClient(url=INFLUXDB_URL, token=INFLUXDB_TOKEN, org=INFLUXDB_ORG)

    write_api = client.write_api(write_options=SYNCHRONOUS)

    write_api.write(
        bucket=INFLUXDB_BUCKET_NAME,
        record=ohlcv,
        data_frame_measurement_name=INFLUXDB_MEASUREMENT_NAME,
        data_frame_timestamp_column="Date",
    )

    client.close()