Spaces:
Runtime error
Runtime error
from typing import List | |
from meta.config import BINANCE_BASE_URL | |
from meta.config import TIME_ZONE_BERLIN | |
from meta.config import TIME_ZONE_JAKARTA | |
from meta.config import TIME_ZONE_PARIS | |
from meta.config import TIME_ZONE_SELFDEFINED | |
from meta.config import TIME_ZONE_SHANGHAI | |
from meta.config import TIME_ZONE_USEASTERN | |
from meta.config import USE_TIME_ZONE_SELFDEFINED | |
from meta.data_processors._base import _Base | |
# from basic_processor import _Base | |
## The code of this file is used in website, not locally. | |
class Quantconnect(_Base): | |
def __init__( | |
self, | |
data_source: str, | |
start_date: str, | |
end_date: str, | |
time_interval: str, | |
**kwargs, | |
): | |
super().__init__(data_source, start_date, end_date, time_interval, **kwargs) | |
# def data_fetch(start_time, end_time, stock_list, resolution=Resolution.Daily) : | |
# #resolution: Daily, Hour, Minute, Second | |
# qb = QuantBook() | |
# for stock in stock_list: | |
# qb.AddEquity(stock) | |
# history = qb.History(qb.Securities.Keys, start_time, end_time, resolution) | |
# return history | |
def download_data( | |
self, ticker_list: List[str], save_path: str = "./data/dataset.csv" | |
): | |
# self.time_zone = calc_time_zone(ticker_list, TIME_ZONE_SELFDEFINED, USE_TIME_ZONE_SELFDEFINED) | |
# start_date = pd.Timestamp(start_date, tz=self.time_zone) | |
# end_date = pd.Timestamp(end_date, tz=self.time_zone) + pd.Timedelta(days=1) | |
qb = QuantBook() | |
for stock in ticker_list: | |
qb.AddEquity(stock) | |
history = qb.History( | |
qb.Securities.Keys, | |
self.start_date, | |
self.end_date, | |
self.time_interval, | |
) | |
self.dataframe = history | |
self.save_data(save_path) | |
print( | |
f"Download complete! Dataset saved to {save_path}. \nShape of DataFrame: {self.dataframe.shape}" | |
) | |
# def preprocess(df, stock_list): | |
# df = df[['open','high','low','close','volume']] | |
# if_first_time = True | |
# for stock in stock_list: | |
# if if_first_time: | |
# ary = df.loc[stock].values | |
# if_first_time = False | |
# else: | |
# temp = df.loc[stock].values | |
# ary = np.hstack((ary,temp)) | |
# return ary | |