mirror of https://github.com/01-edu/public.git
nprimo
2 weeks ago
3 changed files with 88 additions and 30 deletions
@ -1,40 +1,47 @@
|
||||
import os |
||||
import csv |
||||
from datetime import datetime |
||||
|
||||
import pandas as pd |
||||
|
||||
|
||||
def load_historical_data(directory_path='./sample-stocks/'): |
||||
def load_data(directory_path='./sample-stocks/'): |
||||
historical_data = {} |
||||
|
||||
file_list = [filename for filename in os.listdir( |
||||
directory_path) if filename.endswith(".csv")] |
||||
for filename in file_list: |
||||
symbol = filename.replace(".csv", "") |
||||
|
||||
historical_data[symbol] = {} |
||||
file_path = os.path.join(directory_path, filename) |
||||
with open(file_path, 'r') as csv_file: |
||||
csv_reader = csv.DictReader(csv_file) |
||||
historical_data[symbol] = [row['Close'] for row in csv_reader] |
||||
|
||||
historical_data[symbol] = pd.read_csv(file_path, parse_dates=[0]) |
||||
return historical_data |
||||
|
||||
|
||||
def load_data(directory_path='./sample-stocks/'): |
||||
historical_data = {} |
||||
def get_historical_data(df, start, end, start_time): |
||||
today = datetime.fromtimestamp(start_time).date() |
||||
last_entry = df.sort_values(by="Date").iloc[-1] |
||||
delta_today_last_entry = today - last_entry.Date.date() |
||||
|
||||
file_list = [filename for filename in os.listdir( |
||||
directory_path) if filename.endswith(".csv")] |
||||
for filename in file_list: |
||||
symbol = filename.replace(".csv", "") |
||||
file_path = os.path.join(directory_path, filename) |
||||
historical_data[symbol] = pd.read_csv( |
||||
file_path, index_col=0, parse_dates=True) |
||||
try: |
||||
query_start_dt = datetime.fromisoformat(start) |
||||
query_end_dt = datetime.fromisoformat(end) |
||||
if query_end_dt < query_start_dt: |
||||
raise Exception("end_date must come after start_date") |
||||
if query_end_dt.date() > today: |
||||
query_end_dt = datetime.fromtimestamp(start_time) |
||||
|
||||
return historical_data |
||||
df['datetime'] = df.Date + delta_today_last_entry |
||||
return (df.loc[ |
||||
(df.datetime >= query_start_dt) & |
||||
(df.datetime <= query_end_dt)]) |
||||
except Exception as e: |
||||
raise Exception(str(e)) |
||||
|
||||
|
||||
if __name__ == "__main__": |
||||
|
||||
result = load_data() |
||||
print(f'keys: {result.keys()}') |
||||
print(result["AE"].Close[-1]) |
||||
now = datetime.now() |
||||
|
||||
df = result["AE"] |
||||
print(df.info()) |
||||
|
Loading…
Reference in new issue