feat: enhance run_pipeline with error handling and improve logging for data collection process

app.py

@@ -15,6 +15,7 @@ import os
 import time
 import logging
 import json
+import traceback
 from urllib.request import Request, urlopen
 
 # 로깅 설정
@@ -422,7 +423,15 @@ def filter_last_30_days(df):
         return df
 
     df_copy = df.copy()
-    df_copy["_date_parsed"] = pd.to_datetime(df_copy["Date"])
+    df_copy["_date_parsed"] = pd.to_datetime(df_copy["Date"], errors="coerce")
+
+    invalid_date_count = int(df_copy["_date_parsed"].isna().sum())
+    if invalid_date_count > 0:
+        logger.warning(f"Date 파싱 실패 행 {invalid_date_count}개는 30일 필터에서 제외됩니다.")
+
+    df_copy = df_copy[df_copy["_date_parsed"].notna()].copy()
+    if df_copy.empty:
+        return pd.DataFrame(columns=df.columns)
 
     max_date_by_ticker = df_copy.groupby("Ticker")["_date_parsed"].transform("max")
     cutoff_by_ticker = max_date_by_ticker - pd.Timedelta(days=30)
@@ -472,99 +481,110 @@ def run_pipeline(
         return "❌ 허깅페이스 토큰이 필요합니다. HF_TOKEN 환경변수 또는 입력창에 토큰을 넣어주세요."
 
     logs = []
-    logs.append("=" * 60)
-    logs.append("📊 주식 데이터 수집 파이프라인 시작")
-    logs.append(f"⏰ 시작 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
-    logs.append("=" * 60)
-
-    # ========== 1단계: 티커 목록 수집 ==========
-    progress(0, desc="나스닥 & 뉴욕 티커 목록 수집 중...")
-    logs.append("\n🔍 [1단계] 나스닥 & 뉴욕증권거래소 티커 목록 수집 중...")
-
-    nasdaq_tickers, nyse_tickers, all_tickers = get_all_us_tickers()
-
-    logs.append(f"  - 나스닥: {len(nasdaq_tickers)}개")
-    logs.append(f"  - 뉴욕증권거래소: {len(nyse_tickers)}개")
-    logs.append(f"  - 전체: {len(all_tickers)}개")
-
-    if not all_tickers:
-        return "\n".join(logs) + "\n\n❌ 티커 목록을 가져올 수 없습니다."
-
-    # ========== 2단계: 야후 파이낸스 데이터 수집 ==========
-    logs.append(f"\n📥 [2단계] 야후 파이낸스 데이터 수집 시작 (총 {len(all_tickers)}개 티커)")
-    logs.append(f"  - 배치 크기: {batch_size}")
-    logs.append(f"  - 조회 기간(period): {period}")
-    logs.append(f"  ⚠️ 반복문이라 오래 걸립니다. 전체 티커 수에 따라 수 시간 소요될 수 있습니다.")
-
-    all_data_frames = []
-    success_count = 0
-    fail_count = 0
-    total = len(all_tickers)
-
-    for i, ticker in enumerate(all_tickers):
-        # 진행률 업데이트
-        progress_pct = (i + 1) / total
-        progress(progress_pct, desc=f"수집 중: {ticker} ({i + 1}/{total})")
-
-        ticker_df = fetch_ticker_data(ticker, period=period)
-
-        if ticker_df is not None and not ticker_df.empty:
-            all_data_frames.append(ticker_df)
-            success_count += 1
-        else:
-            fail_count += 1
-
-        # 배치 단위로 로그 출력
-        if (i + 1) % batch_size == 0 or (i + 1) == total:
-            logs.append(f"  진행: {i + 1}/{total} (성공: {success_count}, 실패: {fail_count})")
-
-        # API 호출 간 짧은 대기 (야후 차단 방지)
-        if (i + 1) % 10 == 0:
-            time.sleep(0.5)
-
-    logs.append(f"\n📊 수집 완료: 성공 {success_count}개 / 실패 {fail_count}개")
-
-    if not all_data_frames:
-        return "\n".join(logs) + "\n\n❌ 수집된 데이터가 없습니다."
-
-    # ========== 3단계: 데이터 합치기 ==========
-    progress(0.9, desc="데이터 병합 중...")
-    logs.append("\n🔧 [3단계] 티커별 범위 데이터 병합 중...")
-
-    all_df = pd.concat(all_data_frames, ignore_index=True)
-    logs.append(f"  - 전체 데이터: {len(all_df)}행 x {len(all_df.columns)}열")
-    logs.append(f"  - 고유 티커 수: {all_df['Ticker'].nunique()}")
-
-    # ========== 4단계: 30일 필터링 ==========
-    progress(0.93, desc="최근 30일 데이터 필터링 중...")
-    logs.append("\n🗓️ [4단계] 티커별 최근 30일 데이터 필터링 중...")
-
-    recent_30d_df = filter_last_30_days(all_df)
-    progress(0.96, desc="최근 30일 필터링 완료")
-    logs.append(f"  - 30일 데이터: {len(recent_30d_df)}행 x {len(recent_30d_df.columns)}열")
-    logs.append(f"  - 고유 티커 수: {recent_30d_df['Ticker'].nunique()}")
-
-    # ========== 5단계: 허깅페이스 데이터셋 업로드 ==========
-    progress(0.97, desc="all 데이터셋 업로드 중...")
-    logs.append("\n🚀 [5단계] 허깅페이스 데이터셋 업로드 중...")
-
-    # all 데이터셋 업로드
-    result_all = upload_dataset_to_hf(all_df, all_dataset_name, hf_token)
-    logs.append(f"  {result_all}")
-
-    # 30일 데이터셋 업로드
-    progress(0.99, desc="30d 데이터셋 업로드 중...")
-    result_30d = upload_dataset_to_hf(recent_30d_df, recent_dataset_name, hf_token)
-    logs.append(f"  {result_30d}")
+    try:
+        logs.append("=" * 60)
+        logs.append("📊 주식 데이터 수집 파이프라인 시작")
+        logs.append(f"⏰ 시작 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+        logs.append("=" * 60)
+
+        # ========== 1단계: 티커 목록 수집 ==========
+        progress(0, desc="나스닥 & 뉴욕 티커 목록 수집 중...")
+        logs.append("\n🔍 [1단계] 나스닥 & 뉴욕증권거래소 티커 목록 수집 중...")
+
+        nasdaq_tickers, nyse_tickers, all_tickers = get_all_us_tickers()
+
+        logs.append(f"  - 나스닥: {len(nasdaq_tickers)}개")
+        logs.append(f"  - 뉴욕증권거래소: {len(nyse_tickers)}개")
+        logs.append(f"  - 전체: {len(all_tickers)}개")
+
+        if not all_tickers:
+            return "\n".join(logs) + "\n\n❌ 티커 목록을 가져올 수 없습니다."
+
+        # ========== 2단계: 야후 파이낸스 데이터 수집 ==========
+        logs.append(f"\n📥 [2단계] 야후 파이낸스 데이터 수집 시작 (총 {len(all_tickers)}개 티커)")
+        logs.append(f"  - 배치 크기: {batch_size}")
+        logs.append(f"  - 조회 기간(period): {period}")
+        logs.append(f"  ⚠️ 반복문이라 오래 걸립니다. 전체 티커 수에 따라 수 시간 소요될 수 있습니다.")
+
+        all_data_frames = []
+        success_count = 0
+        fail_count = 0
+        total = len(all_tickers)
+
+        for i, ticker in enumerate(all_tickers):
+            # 진행률 업데이트
+            progress_pct = (i + 1) / total
+            progress(progress_pct, desc=f"수집 중: {ticker} ({i + 1}/{total})")
+
+            ticker_df = fetch_ticker_data(ticker, period=period)
+
+            if ticker_df is not None and not ticker_df.empty:
+                all_data_frames.append(ticker_df)
+                success_count += 1
+            else:
+                fail_count += 1
+
+            # 배치 단위로 로그 출력
+            if (i + 1) % batch_size == 0 or (i + 1) == total:
+                logs.append(f"  진행: {i + 1}/{total} (성공: {success_count}, 실패: {fail_count})")
+
+            # API 호출 간 짧은 대기 (야후 차단 방지)
+            if (i + 1) % 10 == 0:
+                time.sleep(0.5)
+
+        logs.append(f"\n📊 수집 완료: 성공 {success_count}개 / 실패 {fail_count}개")
+
+        if not all_data_frames:
+            return "\n".join(logs) + "\n\n❌ 수집된 데이터가 없습니다."
+
+        # ========== 3단계: 데이터 합치기 ==========
+        progress(0.9, desc="데이터 병합 중...")
+        logs.append("\n🔧 [3단계] 티커별 범위 데이터 병합 중...")
+
+        all_df = pd.concat(all_data_frames, ignore_index=True)
+        logs.append(f"  - 전체 데이터: {len(all_df)}행 x {len(all_df.columns)}열")
+        logs.append(f"  - 고유 티커 수: {all_df['Ticker'].nunique()}")
+
+        # ========== 4단계: 30일 필터링 ==========
+        progress(0.93, desc="최근 30일 데이터 필터링 중...")
+        logs.append("\n🗓️ [4단계] 티커별 최근 30일 데이터 필터링 중...")
+
+        recent_30d_df = filter_last_30_days(all_df)
+        progress(0.96, desc="최근 30일 필터링 완료")
+        logs.append(f"  - 30일 데이터: {len(recent_30d_df)}행 x {len(recent_30d_df.columns)}열")
+        logs.append(f"  - 고유 티커 수: {recent_30d_df['Ticker'].nunique()}")
+
+        # ========== 5단계: 허깅페이스 데이터셋 업로드 ==========
+        progress(0.97, desc="all 데이터셋 업로드 중...")
+        logs.append("\n🚀 [5단계] 허깅페이스 데이터셋 업로드 중...")
+
+        # all 데이터셋 업로드
+        result_all = upload_dataset_to_hf(all_df, all_dataset_name, hf_token)
+        logs.append(f"  {result_all}")
+
+        # 30일 데이터셋 업로드
+        progress(0.99, desc="30d 데이터셋 업로드 중...")
+        result_30d = upload_dataset_to_hf(recent_30d_df, recent_dataset_name, hf_token)
+        logs.append(f"  {result_30d}")
+
+        # ========== 완료 ==========
+        progress(1.0, desc="완료!")
+        logs.append("\n" + "=" * 60)
+        logs.append(f"✅ 파이프라인 완료!")
+        logs.append(f"⏰ 종료 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+        logs.append("=" * 60)
 
-    # ========== 완료 ==========
-    progress(1.0, desc="완료!")
-    logs.append("\n" + "=" * 60)
-    logs.append(f"✅ 파이프라인 완료!")
-    logs.append(f"⏰ 종료 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
-    logs.append("=" * 60)
+        return "\n".join(logs)
 
-    return "\n".join(logs)
+    except Exception as e:
+        logger.exception("run_pipeline 실행 중 예외 발생")
+        logs.append("\n" + "=" * 60)
+        logs.append("❌ 파이프라인 실행 중 예외가 발생했습니다.")
+        logs.append(f"오류 메시지: {e}")
+        logs.append("\n[Traceback]")
+        logs.append(traceback.format_exc())
+        logs.append("=" * 60)
+        return "\n".join(logs)
 
 
 def preview_tickers():