Spaces:
Sleeping
Sleeping
| #-------------------------------------libraries ---------------------------------- | |
| from requests import Request, Session | |
| from requests.exceptions import ConnectionError, Timeout, TooManyRedirects | |
| import json | |
| import os | |
| import pandas as pd | |
| import numpy as np | |
| import logging | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| #-------------------------------------env vars---------------------------------- | |
| url = os.getenv("URL_CMC") | |
| endpoints = ["v1/cryptocurrency/listings/latest", | |
| "/v1/cryptocurrency/trending/latest", | |
| ] | |
| start = "1" | |
| stop = "100" | |
| parameters = { | |
| 'start':start, | |
| 'limit':stop, | |
| 'convert':'USD' | |
| } | |
| headers = { | |
| 'Accepts': 'application/json', | |
| 'X-CMC_PRO_API_KEY': os.getenv("API_KEY_CMC"), | |
| } | |
| # Configure the logging settings | |
| log_folder = "./logs/scrapping/" | |
| os.makedirs(log_folder, exist_ok=True) # Ensure the log folder exists | |
| log_file = os.path.join(log_folder, "scrapping.log") | |
| log_format = "%(asctime)s [%(levelname)s] - %(message)s" | |
| logging.basicConfig(filename=log_file, level=logging.INFO, format=log_format) | |
| #-------------------------------------api call---------------------------------- | |
| session = Session() | |
| session.headers.update(headers) | |
| for endpoint in endpoints: | |
| target = f"{url}/{endpoint}" | |
| try: | |
| response = session.get(target, params=parameters) | |
| data = json.loads(response.text) | |
| with open(f'output/cmc_data_{endpoint.replace("/", "_")}_{stop}.json', 'w') as f: | |
| json.dump(data, f) | |
| logging.info(f"Successfully fetched data from {target}") | |
| except (ConnectionError, Timeout, TooManyRedirects) as e: | |
| logging.error(f"Error while fetching data from {target}: {e}") | |
| #-------------------------------------process data---------------------------------- | |
| # create data frame with chosen columns | |
| df = pd.DataFrame(data["data"])[["name","symbol","circulating_supply","total_supply","quote"]] | |
| # explode column quote then chose columns | |
| quote_df = pd.json_normalize(df['quote'].apply(lambda x: x['USD']))[["price","percent_change_24h","percent_change_7d","percent_change_90d","market_cap","fully_diluted_market_cap","last_updated"]] | |
| # drop quote | |
| df = df.drop("quote",axis=1) | |
| # create features | |
| df["percent_tokens_circulation"] = np.round((df["circulating_supply"]/df["total_supply"])*100,1) | |
| # merge dataframe | |
| df = df.join(quote_df) | |
| df["last_updated"] = pd.to_datetime(df["last_updated"]) | |
| #df.to_csv(f"output/top_{stop}_update.csv") | |
| #-------------------------------------save data---------------------------------- | |
| # Check if the file exists | |
| output_file = f"output/top_{stop}_update.csv" | |
| if os.path.isfile(output_file): | |
| logging.info("Updating dataset"+f"top_{stop}_update"+". ") | |
| # Read the existing data | |
| existing_data = pd.read_csv(output_file) | |
| # Concatenate the existing data with the new data vertically | |
| updated_data = pd.concat([existing_data, df], axis=0, ignore_index=True) | |
| # Remove duplicates (if any) based on a unique identifier column | |
| updated_data.drop_duplicates(subset=["symbol", "last_updated"], inplace=True) | |
| # Save the updated data back to the same file | |
| updated_data.to_csv(output_file, index=False) | |
| else: | |
| # If the file doesn't exist, save the current data to it | |
| df.to_csv(output_file, index=False) | |
| logging.info("Script execution completed.") | |
| #-------------------------------------end---------------------------------- | |