Skip to content

Instantly share code, notes, and snippets.

@Jordach
Last active September 1, 2022 01:55
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save Jordach/912e15c4585f2030fb343c4ad4c0beed to your computer and use it in GitHub Desktop.
Save Jordach/912e15c4585f2030fb343c4ad4c0beed to your computer and use it in GitHub Desktop.
from datetime import datetime, date, timezone
import numpy as np
import pandas as pd
import sqlite3
def get_stock_from_db(ticker, interval, limit=-1):
# Translate between local SQL db and expected formatting:
pwd = os.getcwd()
name = ticker.lower()
con = sqlite3.connect(pwd + "/db/db_" + name + ".db")
cur = con.cursor()
ohlc_data = []
# Create an SQL command to get data
query = "SELECT * FROM " + interval.lower() + " ORDER BY date DESC"
if limit > 0:
query += " LIMIT " + str(limit)
else:
query += " LIMIT 4000"
# Grab data in rows
for row in cur.execute(query):
ohlc_data.append(row)
# Fix inverse sorting - ie making the earliest entry index 0
ohlc_data.sort(key=lambda tup: tup[0])
dt = {"date":[], "Open":[], "High":[], "Low":[], "Close":[], "sma":[]}
if interval == "m1":
for item in ohlc_data:
dt["date"].append(datetime.utcfromtimestamp(int(item[0])).strftime('%H:%M:%S %d/%m/%y'))
dt["Open"].append(float(item[1]))
dt["High"].append(float(item[1]))
dt["Low"].append(float(item[1]))
dt["Close"].append(float(item[1]))
dt["sma"].append(float(item[1]))
else:
for item in ohlc_data:
dt["date"].append(datetime.utcfromtimestamp(int(item[0])).strftime('%H:%M:%S %d/%m/%y'))
dt["Open"].append(float(item[1]))
dt["High"].append(float(item[2]))
dt["Low"].append(float(item[3]))
dt["Close"].append(float(item[4]))
dt["sma"].append(float(item[4]))
return dt
your_stock_dataframe = get_stock_from_db("sym", "d1" 30)
print(your_stock_dataframe)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment