This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
def get_stat_file(stat_code: str) -> pd.DataFrame: | |
""" | |
Downloads and extracts the statistics table from Statistics Canada. | |
stat_code (str) : the statistics Canada code for the needed table. | |
Returns: | |
A Pandas DataFrame containing the columns of the requested statitics | |
""" | |
url = 'https://www150.statcan.gc.ca/t1/wds/rest/getFullTableDownloadCSV/' + stat_code + '/en' |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import requests | |
import pandas as pd | |
import json | |
from zipfile import ZipFile | |
# Fetch the primary data about cubes of data | |
url = "https://www150.statcan.gc.ca/t1/wds/rest/getAllCubesList" | |
response = requests.get(url) | |
cubes = [{key:val for key, val in itm.items() if key in ["productId", "cubeEndDate", "dimensions"]} for itm in response.json()] | |
cubes_df = pd.DataFrame(cubes) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
def dc(ohlcv, thresh=0.005): | |
upturn_event = True | |
p_h = p_l = ohlcv['close'][0] | |
dc_ranges = defaultdict(list) | |
tuples = tuple(ohlcv.itertuples()) | |
# here we find the timedelta between the timeseries index, 1 and 0, so 1 day for daily data | |
step = tuples[1].Index - tuples[0].Index | |
# loop over tuples of ohlcv and time as Index, |