PyAirbyte Custom Snowflake Cache Demo

# Add virtual environment support for running in Google Colab
!apt-get install -qq python3.10-venv

# Install PyAirbyte
%pip install --quiet airbyte

from airbyte.caches import SnowflakeCache
from google.colab import userdata

# Define a Snowflake Cache and pass the necessary configuration
sf_cache = SnowflakeCache(
      account=userdata.get("SNOWFLAKE_ACCOUNT"),
      username=userdata.get("SNOWFLAKE_USERNAME"),
      password=userdata.get("SNOWFLAKE_PASSWORD"),
      warehouse="AIRBYTE_DEVELOP_WAREHOUSE",
      database="AIRBYTE_DEVELOP",
      role="AIRBYTE_DEVELOPER",
      schema_name="PYAIRBYTE_DEMO"
)

import airbyte as ab

# Configure and read from the source
read_result = ab.get_source(
    "source-coin-api",
    config={
          "api_key": userdata.get("API_KEY"),
          "environment": "production",
          "symbol_id": "COINBASE_SPOT_INDEX_USD",
          "period": "1DAY",
          "start_date": "2023-01-01T00:00:00"
    },
    streams=["ohlcv_historical_data", "trades_historical_data", "quotes_historical_data"],
).read(cache=sf_cache)

# Read from the cache into a pandas Dataframe:
ohlcv_df = read_result["ohlcv_historical_data"].to_pandas()

import pandas as pd

# Convert 'time_period_start' to datetime format and necessary columns to numeric
ohlcv_df['time_period_start'] = pd.to_datetime(ohlcv_df['time_period_start'])
numeric_columns = ['price_open', 'price_high', 'price_low', 'price_close', 'volume_traded', 'trades_count']
ohlcv_df[numeric_columns] = ohlcv_df[numeric_columns].apply(pd.to_numeric, errors='coerce')

# Calculate daily price movement
ohlcv_df['daily_movement'] = ohlcv_df['price_close'] - ohlcv_df['price_open']

engine = sf_cache.get_sql_engine()

from snowflake.connector.pandas_tools import pd_writer

ohlcv_df.to_sql('daily_movement', engine, index=False, method=pd_writer, if_exists='replace')