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

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 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 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 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 run_indicator(self):
    if self.indicator == 'bbands':
        self.setup_bbands()
    elif self.indicator == 'sma':
        self.setup_sma()

natyrix

class VectorbotPipeline():
    def __init__(self,user_id, indicator="sma",init_cash=1000, ema_value=0,stock='AMZN', fast_ma=10, slow_ma=50, start='2021-10-11', end='2022-10-11', period=None, fees=0.005, is_experiment=False):
        self.user_id = user_id
        self.init_cash = init_cash
        self.ema_value = ema_value
        self.indicator = indicator
        self.stock = stock
        self.fast_ma = fast_ma
        self.slow_ma = slow_ma
        self.start = start

natyrix

import unittest
import pandas as pd
import sys
import os

sys.path.append(os.path.abspath(os.path.join("../Twitter-Data-Analysis/")))

from extract_dataframe import read_json
from extract_dataframe import TweetDfExtractor
from clean_tweets_dataframe import Clean_Tweets

natyrix

import numpy as np
import pandas as pd
import streamlit as st
import altair as alt
from wordcloud import WordCloud
import plotly.express as px
from textblob import TextBlob
from sklearn.feature_extraction.text import CountVectorizer, TfidfTransformer
import pickle

natyrix

import numpy as np
import pandas as pd
import streamlit as st
import altair as alt
from wordcloud import WordCloud
import plotly.express as px 
from textblob import TextBlob
from sklearn.feature_extraction.text import CountVectorizer, TfidfTransformer
import pickle
from  pages.plots import *