feat: implement financial market data DAG and add new automated data updater DAGs for technical indicators, funding rates, and news.

Author: mouadja02

calculate_technical_indicators.py

@@ -12,7 +12,7 @@
     python calculate_technical_indicators.py
 
 Input:
-    - CSV file with columns: UNIX_TIMESTAMP, DATETIME, OPEN, HIGH, CLOSE, LOW, VOLUME
+    - CSV file with columns: UNIX_TIMESTAMP, DATETIME, OPEN, HIGH, CLOSE, LOW, VOLUME (or VOLUME_USD)
 
 Output:
     - CSV file with OHLCV + 90+ technical indicators
@@ -24,15 +24,8 @@
 
 
 def load_ohlcv_data(filepath='bitcoin-hourly-ohlcv.csv'):
-    """
-    Load OHLCV data from CSV file
-    
-    Args:
-        filepath: Path to CSV file with OHLCV data
-        
-    Returns:
-        pandas DataFrame with OHLCV data
-    """
+    # Load OHLCV data from CSV file
+    
     print(f"📥 Loading data from {filepath}...")
     df = pd.read_csv(filepath)
 
@@ -45,15 +38,8 @@ def load_ohlcv_data(filepath='bitcoin-hourly-ohlcv.csv'):
 
 
 def calculate_technical_indicators(df):
-    """
-    Calculate comprehensive technical indicators using TA-Lib
-    
-    Args:
-        df: DataFrame with OHLCV data (columns: OPEN, HIGH, CLOSE, LOW, VOLUME)
-        
-    Returns:
-        DataFrame with all original columns + technical indicators
-    """
+    # Calculate comprehensive technical indicators using TA-Lib
+
     print("🔧 Calculating comprehensive technical indicators using TA-Lib...")
 
     # Convert to numpy arrays for TA-Lib
@@ -252,27 +238,13 @@ def calculate_technical_indicators(df):
 
 
 def save_to_csv(df, filepath='bitcoin-hourly-technical-indicators.csv'):
-    """
-    Save DataFrame to CSV file
-    
-    Args:
-        df: DataFrame to save
-        filepath: Output file path
-    """
+    # Save DataFrame to CSV file
     print(f"💾 Saving to {filepath}...")
     df.to_csv(filepath, index=False, float_format='%.8f')
     print(f"✅ Saved {len(df):,} records to {filepath}")
 
 
-def main():
-    """
-    Main function to run the technical indicators calculation
-    """
-    print("=" * 60)
-    print("Bitcoin Technical Indicators Calculator")
-    print("=" * 60)
-    print()
-    
+def main():  
     # Step 1: Load OHLCV data
     df = load_ohlcv_data('bitcoin-hourly-ohlcv.csv')
     print()
@@ -284,20 +256,6 @@ def main():
     # Step 3: Save to CSV
     save_to_csv(df_with_indicators, 'bitcoin-hourly-technical-indicators.csv')
     print()
-    
-    print("=" * 60)
-    print("✅ Process Complete!")
-    print("=" * 60)
-    print()
-    print("📁 Output Files:")
-    print("  • bitcoin-hourly-technical-indicators.csv")
-    print()
-    print("📊 Summary:")
-    print(f"  • Total Records: {len(df_with_indicators):,}")
-    print(f"  • Total Columns: {len(df_with_indicators.columns)}")
-    print(f"  • OHLCV Columns: 7")
-    print(f"  • Technical Indicators: {len(df_with_indicators.columns) - 7}")
-    print()
 
 
 if __name__ == "__main__":

dags/OpenInterestFutures_data_updater_dag.py

@@ -48,6 +48,48 @@
     tags=['bitcoin', 'futures', 'open-interest', 'daily', 'snowflake']
 )
 
+def ensure_schema_and_table(**context):
+    """Create database, schema, stage and table if they don't exist"""
+    
+    snowflake_hook = SnowflakeHook(snowflake_conn_id='snowflake_default')
+    
+    # Create database if not exists
+    snowflake_hook.run("CREATE DATABASE IF NOT EXISTS BITCOIN_DATA")
+    print("✅ Database BITCOIN_DATA ensured")
+    
+    # Create schemas if not exists
+    snowflake_hook.run("CREATE SCHEMA IF NOT EXISTS BITCOIN_DATA.DATA")
+    snowflake_hook.run("CREATE SCHEMA IF NOT EXISTS BITCOIN_DATA.FORECASTER")
+    print("✅ Schemas ensured")
+    
+    # Create stage and file format
+    snowflake_hook.run("""
+        CREATE FILE FORMAT IF NOT EXISTS BITCOIN_DATA.FORECASTER.json_format
+        TYPE = 'JSON' STRIP_OUTER_ARRAY = TRUE
+    """)
+    snowflake_hook.run("""
+        CREATE STAGE IF NOT EXISTS BITCOIN_DATA.FORECASTER.my_stage
+        FILE_FORMAT = BITCOIN_DATA.FORECASTER.json_format
+    """)
+    print("✅ Stage and file format ensured")
+    
+    # Create table if not exists
+    snowflake_hook.run("""
+    CREATE TABLE IF NOT EXISTS BITCOIN_DATA.DATA.OPEN_INTEREST_FUTURES (
+        date DATE,
+        unix_ts BIGINT,
+        binance FLOAT, bybit FLOAT, okx FLOAT, bitget FLOAT,
+        deribit FLOAT, bitmex FLOAT, huobi FLOAT, bitfinex FLOAT,
+        gate_io FLOAT, kucoin FLOAT, kraken FLOAT, crypto_com FLOAT,
+        dydx FLOAT, delta_exchange FLOAT, total_open_interest FLOAT,
+        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP(),
+        updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP(),
+        PRIMARY KEY (date)
+    )
+    """)
+    print("✅ Table BITCOIN_DATA.DATA.OPEN_INTEREST_FUTURES ensured")
+
+
 def download_and_upload_open_interest(**context):
     """
     Download open futures interest JSON data from API and upload to Snowflake stage
@@ -237,6 +279,13 @@ def cleanup_stage_files(**context):
 
     return result
 
+# Ensure DB infrastructure exists
+ensure_db_task = PythonOperator(
+    task_id='ensure_schema_and_table',
+    python_callable=ensure_schema_and_table,
+    dag=dag
+)
+
 # Create file format task
 create_file_format = SnowflakeOperator(
     task_id='create_file_format',
@@ -271,4 +320,5 @@ def cleanup_stage_files(**context):
 )
 
 # Set task dependencies
-create_file_format >> download_task >> merge_task >> cleanup_task
+ensure_db_task >> create_file_format >> download_task >> merge_task >> cleanup_task
+

dags/bitcoin_news_dag.py

@@ -1,6 +1,7 @@
 """
 Bitcoin News DAG - Production Version
 Fetches Bitcoin news from multiple APIs and stores in Snowflake with duplicate prevention
+Includes database initialization with schema/table creation for data recovery
 """
 
 from datetime import datetime, timedelta
@@ -35,6 +36,42 @@
     tags=['bitcoin', 'news', 'snowflake', 'production'],
 )
 
+# ─── Database Initialization ───────────────────────────────────────────
+
+def ensure_schema_and_table(**context):
+    """Create database, schema and table in Snowflake if they don't exist"""
+    
+    hook = SnowflakeHook(snowflake_conn_id='snowflake_default')
+    
+    # Create database if not exists
+    hook.run("CREATE DATABASE IF NOT EXISTS BITCOIN_DATA")
+    print("✅ Database BITCOIN_DATA ensured")
+    
+    # Create RAW schema if not exists (news goes to RAW schema)
+    hook.run("CREATE SCHEMA IF NOT EXISTS BITCOIN_DATA.RAW")
+    print("✅ Schema BITCOIN_DATA.RAW ensured")
+    
+    # Create table if not exists
+    create_table_sql = """
+    CREATE TABLE IF NOT EXISTS BITCOIN_DATA.RAW.BITCOIN_NEWS (
+        datetime TIMESTAMP,
+        headline VARCHAR(2000),
+        summary VARCHAR(10000),
+        source VARCHAR(500),
+        url VARCHAR(1000),
+        categories VARCHAR(1000),
+        tags VARCHAR(1000),
+        api_source VARCHAR(100),
+        file_name VARCHAR(500),
+        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
+    )
+    """
+    hook.run(create_table_sql)
+    print("✅ Table BITCOIN_DATA.RAW.BITCOIN_NEWS ensured")
+
+
+# ─── Data Functions ────────────────────────────────────────────────────
+
 def get_last_datetime_from_snowflake():
     """Get the latest datetime from Snowflake to avoid duplicates"""
 
@@ -399,7 +436,15 @@ def send_production_notification(**context):
     except Exception as e:
         print(f"Failed to send notification: {str(e)}")
 
-# Define tasks
+# ─── Task Definitions ──────────────────────────────────────────────────
+
+# Step 1: Ensure DB infrastructure exists
+ensure_db_task = PythonOperator(
+    task_id='ensure_schema_and_table',
+    python_callable=ensure_schema_and_table,
+    dag=dag,
+)
+
 fetch_cryptocompare_task = PythonOperator(
     task_id='fetch_cryptocompare_news',
     python_callable=fetch_cryptocompare_news,
@@ -436,5 +481,7 @@ def send_production_notification(**context):
     dag=dag,
 )
 
-# Set task dependencies
-[fetch_cryptocompare_task, fetch_finnhub_task] >> merge_deduplicate_task >> insert_snowflake_task >> notification_task
+# ─── Task Dependencies ─────────────────────────────────────────────────
+# ensure_schema_and_table >> [fetch_*] >> merge >> insert >> notification
+
+ensure_db_task >> [fetch_cryptocompare_task, fetch_finnhub_task] >> merge_deduplicate_task >> insert_snowflake_task >> notification_task