espin086 / espin086/AppGenie

Create Class to Get FRED Data

Open
#7 0 comments 0 reactions 1 assignee Claimed by @espin086 View on GitHub
enhancement
Dominant language
Python
Stars
1
Forks
1
PR merge metrics
No merged PRs in 30d

Description

here is some sample code:

```
import logging
from argparse import ArgumentParser
import pandas as pd
from fredapi import Fred
from bigqueryhandler import BigQueryHandler
from csvhandler import CSVHandler
from sqlitecrud import SQLiteCRUD
from excel import ExcelHandler

__author__ = 'Your Name'
__version__ = '1.0.0'
__maintainer__ = 'Your Name'
__email__ = 'your.email@example.com'

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

class FREDDataCollector:
"""
A class to collect and store macroeconomic data from FRED API.

Attributes:
api_key (str): The API key for the FRED service.
series_ids (list): List of FRED series IDs to collect.
save_to (str): The storage format ('csv', 'sqlite', 'excel').
output_path (str): Path to the output file or database.
"""

def __init__(self, api_key: str, series_ids: list, save_to: str, output_path: str) -> None:
"""
Initializes the FREDDataCollector with relevant parameters.

Args:
api_key (str): The API key for the FRED service.
series_ids (list): List of FRED series IDs to collect.
save_to (str): The storage format ('csv', 'sqlite', 'excel').
output_path (str): Path to the output file or database.
"""
assert isinstance(api_key, str), "api_key must be a string."
assert isinstance(series_ids, list), "series_ids must be a list."
assert isinstance(save_to, str), "save_to must be a string."
assert isinstance(output_path, str), "output_path must be a string."

self.fred = Fred(api_key=api_key)
self.series_ids = series_ids
self.save_to = save_to
self.output_path = output_path
self.data = pd.DataFrame()

logging.info("FRED Data Collector initialized.")

def collect_data(self) -> pd.DataFrame:
"""
Collects data from FRED based on the specified series IDs.

Returns:
pd.DataFrame: DataFrame containing the collected FRED data.
"""
all_series = []
for series_id in self.series_ids:
try:
logging.info(f"Collecting data for series ID: {series_id}")
series_data = self.fred.get_series(series_id)
series_data = series_data.rename(series_id)
all_series.append(series_data)
except Exception as e:
logging.error(f"Error collecting data for series ID {series_id}: {e}")

self.data = pd.concat(all_series, axis=1)
logging.info("Data collection completed.")
return self.data

def save_data(self) -> None:
"""
Saves the collected data based on the specified save_to parameter.
"""
if self.save_to == "csv":
csv_handler = CSVHandler(self.output_path)
csv_handler.save_csv(self.data)
elif self.save_to == "sqlite":
db_handler = SQLiteCRUD(self.output_path)
db_handler.connect()
columns = ', '.join([f'"{col}" TEXT' for col in self.data.columns])
db_handler.create_table("macro_data", columns)
for idx, row in self.data.iterrows():
db_handler.insert_data("macro_data", tuple(row))
db_handler.close()
elif self.save_to == "excel":
excel_handler = ExcelHandler(self.output_path)
excel_handler.save_sheet(self.data, "MacroData")
else:
logging.error(f"Unsupported save_to format: {self.save_to}")

def main():
"""Main function for the FRED Data Collector script."""
parser = ArgumentParser(description="FRED Data Collector")
parser.add_argument("--api_key", type=str, required=True, help="FRED API key.")
parser.add_argument("--series_ids", type=str, nargs='+', required=True, help="FRED series IDs.")
parser.add_argument("--save_to", type=str, required=True, choices=['csv', 'sqlite', 'excel'], help="Storage format.")
parser.add_argument("--output_path", type=str, required=True, help="Path to save the collected data.")
args = parser.parse_args()

try:
collector = FREDDataCollector(args.api_key, args.series_ids, args.save_to, args.output_path)
data = collector.collect_data()
assert not data.empty, "Collected data is empty."
collector.save_data()
logging.info("Data collection and storage completed successfully.")
except Exception as e:
logging.error(f"An error occurred: {e}")

if __name__ == "__main__":
main()
```

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.