natyrix

def run_indicator(self):
    if self.indicator == 'bbands':
        self.setup_bbands()
    elif self.indicator == 'sma':
        self.setup_sma()

natyrix

def setup_sma(self):
    if self.is_experiment:
        mlflow.log_param("stock_sma", self.stock)
        mlflow.log_param("stock_fast_sma", self.fast_ma)
        mlflow.log_param("stock_slow_sma", self.slow_ma)
        mlflow.log_metric("init_cash", self.init_cash)
    price = vbt.YFData.download(self.stock, start=self.start, end=self.end).get('Close')
    self.calc_fast_ma = vbt.MA.run(self.price, self.fast_ma, short_name='fast_ma')
    self.calc_slow_ma = vbt.MA.run(self.price, self.slow_ma, short_name='slow_ma')
    entries = self.calc_fast_ma.ma_crossed_above(self.calc_slow_ma)

natyrix

def setup_bbands(self):
    bbands = vbt.BBANDS.run(self.price)
    entries = bbands.close_crossed_below(bbands.lower)
    exits = bbands.close_crossed_above(bbands.upper)
    self.pf = vbt.Portfolio.from_signals(self.price, entries, exits)

natyrix

def return_backtest_result(self):
    if self.is_experiment:
        mlflow.log_param("stock", self.stock)
        mlflow.log_param("init_cash", self.init_cash)
        for k,v in self.pf.stats().to_dict().items():
            mlflow.log_param(str(k).replace('%','').replace('[','').replace(']',''),str(v))
    with open('./backtest_result/vectorbt/fast_and_slow_plot.txt','w') as f:
        for key, value in self.pf.stats().to_dict().items(): 
            f.write('%s: %s\n' % (key, value))
        # f.write(self.pf.stats().to_dict())

natyrix

def save_result_and_publish(self):
      result_dict = self.return_backtest_result()
      DB_URL = os.environ["DB_URL"]
      sql_engine = create_engine(DB_URL)
      session = Session(sql_engine)

      user = session.query(User).filter_by(id=self.user_id).first()

      if user:
          backtest_scene = BackTestScene(coin_name="BTC",

natyrix

def plot_fast_and_slow(self):
    fig = self.price.vbt.plot(trace_kwargs=dict(name='Close'))
    self.calc_fast_ma.ma.vbt.plot(trace_kwargs=dict(name='Fast MA'), fig=fig)
    self.calc_slow_ma.ma.vbt.plot(trace_kwargs=dict(name='Slow MA'), fig=fig)
    self.pf.positions.plot(close_trace_kwargs=dict(visible=False), fig=fig)
    # vbt.save('fig.png', fig)
    with open('./images/vectorbt/fast_and_slow_plot.png','wb') as f:
        f.write(fig.to_image(format='png'))
    with open('./images/vectorbt/fast_and_slow_plot.png','rb') as f:
        if self.is_experiment:

natyrix

CREATE TABLE "User_Table" (
  "ID" varchar(PK),
  "Password" varchar,
  "Email" varchar,
  "Subscription_ID" varchar(FK)
);

CREATE TABLE "Result_Table" (
  "ID" varchar(PK),
  "Back_test_ID" varcahr(FK),

natyrix

async function runTest(e){
    e.preventDefault()
    try{
        setIsLoading(true)
        let response = await axios.post(`${BASE_URL}/get_backtest_scene`, {
            "start_date":startDate,
            "end_date":endDate,
            "indicator":selectedindicator,
            "initial_cash":initCash,
            "stock":selectedCoin,

natyrix

def isWeekend(self, df_date_str):
      # 2021-07-01 07:28:04
      datetime_object = datetime.strptime(df_date_str, '%Y-%m-%d %H:%M:%S')
      # print(datetime_object.weekday())
      if datetime_object.weekday() < 5:
          return 0
      else:  # 5 Sat, 6 Sun
          return 1

natyrix

import holidays

class EDAPipeline():

    def __init__(self) -> None:
        self.ng_holidays = holidays.country_holidays('NG')
    
    def isHoliday(self, df_date_str):
        try:
            dt = datetime.strptime(df_date_str, '%Y-%m-%d %H:%M:%S').date()