feat: enhance run_pipeline to load existing datasets and calculate pending tickers for collection

README.md

@@ -72,7 +72,10 @@ UI 연결:
 1. **티커 목록 수집**
    - `get_all_us_tickers()` 호출
    - 전체 대상 티커 개수 계산
-2. **티커별 가격 데이터 수집**
+2. **기존 데이터셋 로드 + 재개 대상 계산**
+   - 기존 `all`, `30d` 데이터셋 로드 시도
+   - 이미 수집된 티커를 제외하고 이번 실행 대상 티커만 계산
+3. **티커별 가격 데이터 수집**
    - 각 티커에 대해 `fetch_ticker_data(ticker)` 실행
    - `yf.Ticker(ticker).history(period=period, interval="1d")` 호출
    - 컬럼 정리: `Ticker`, `Date`, `Open`, `High`, `Low`, `Close`, `Volume`
@@ -82,16 +85,17 @@ UI 연결:
      - 마감 확정 전(today) 데이터는 제외
    - 실패 시 최대 3회 재시도
    - 10티커마다 0.5초 대기(호출 차단 완화)
-   - 설정한 간격(`checkpoint_batch_size`)마다 중간 업로드(체크포인트) 수행 가능
-3. **전체 데이터 병합**
-   - 수집 성공한 티커 데이터프레임을 `concat`
-4. **최근 30일 데이터셋 생성**
-   - 티커별로 미리 계산한 30일 데이터들을 병합
-5. **Hugging Face 업로드**
+   - 설정한 간격(`checkpoint_batch_size`)마다 기존 데이터셋과 병합 후 중간 업로드(체크포인트) 수행
+4. **최종 병합 반영**
+   - 마지막 미반영 버퍼를 기존 데이터셋에 병합
+   - `Ticker + Date` 기준 중복 제거(최신값 유지)
+5. **최근 30일 데이터셋 갱신**
+   - 병합된 데이터 기준으로 티커별 30일 윈도우 유지
+6. **Hugging Face 업로드**
    - `upload_dataset_to_hf(all_df, all_dataset_name, hf_token)`
    - `upload_dataset_to_hf(recent_30d_df, recent_dataset_name, hf_token)`
    - 내부적으로 `Dataset.from_pandas(...).push_to_hub(...)` 사용
-6. **완료 로그 반환**
+7. **완료 로그 반환**
    - 시작/종료 시간, 성공/실패 개수, 업로드 결과를 텍스트로 반환
 
 진행률(Progress Bar):
@@ -162,6 +166,11 @@ UI 연결:
   - 예: `100`이면 100개 티커 처리마다 중간 업로드
   - `0`이면 중간 업로드 없이 마지막에 한 번만 업로드
 
+재실행(Resume) 동작:
+
+- 이미 데이터셋에 존재하는 티커는 자동으로 스킵합니다.
+- 이전 실행에서 중간에 실패해도, 다음 실행 시 남은 티커 위주로 이어서 수집합니다.
+
 ---
 
 ## 참고

app.py

@@ -16,6 +16,7 @@ import time
 import logging
 import json
 import traceback
+import gc
 from urllib.request import Request, urlopen
 
 # 로깅 설정
@@ -483,6 +484,16 @@ def run_pipeline(
 
     logs = []
     try:
+        def _merge_rows(base_df, incoming_df):
+            if incoming_df is None or incoming_df.empty:
+                return base_df
+            if base_df is None or base_df.empty:
+                return incoming_df.reset_index(drop=True)
+
+            merged = pd.concat([base_df, incoming_df], ignore_index=True)
+            merged = merged.drop_duplicates(subset=["Ticker", "Date"], keep="last")
+            return merged.reset_index(drop=True)
+
         logs.append("=" * 60)
         logs.append("📊 주식 데이터 수집 파이프라인 시작")
         logs.append(f"⏰ 시작 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
@@ -541,8 +552,41 @@ def run_pipeline(
 
                 return "\n".join(logs) + "\n\n❌ 티커 목록을 가져올 수 없습니다."
 
-        # ========== 2단계: 야후 파이낸스 데이터 수집 ==========
-        logs.append(f"\n📥 [2단계] 야후 파이낸스 데이터 수집 시작 (총 {len(all_tickers)}개 티커)")
+        # ========== 2단계: 기존 데이터셋 로드 + 재개 대상 계산 ==========
+        progress(0.08, desc="기존 데이터셋 로드 중...")
+        logs.append("\n📂 [2단계] 기존 데이터셋 로드 및 재개 대상 계산...")
+
+        base_all_df = pd.DataFrame(columns=["Ticker", "Date", "Open", "High", "Low", "Close", "Volume"])
+        base_recent_df = pd.DataFrame(columns=["Ticker", "Date", "Open", "High", "Low", "Close", "Volume"])
+
+        try:
+            base_all_df = load_hf_dataset_as_df(all_dataset_name, hf_token)
+            logs.append(f"  - 기존 all 데이터: {len(base_all_df)}행")
+        except Exception as e:
+            logs.append(f"  - 기존 all 데이터 로드 실패(신규 생성으로 진행): {e}")
+
+        try:
+            base_recent_df = load_hf_dataset_as_df(recent_dataset_name, hf_token)
+            logs.append(f"  - 기존 30d 데이터: {len(base_recent_df)}행")
+        except Exception as e:
+            logs.append(f"  - 기존 30d 데이터 로드 실패(신규 생성으로 진행): {e}")
+
+        existing_tickers = set(base_all_df["Ticker"].dropna().astype(str).str.upper().tolist())
+        if not existing_tickers:
+            existing_tickers = set(base_recent_df["Ticker"].dropna().astype(str).str.upper().tolist())
+
+        pending_tickers = [ticker for ticker in all_tickers if ticker not in existing_tickers]
+
+        logs.append(f"  - 기존 수집 티커: {len(existing_tickers)}개")
+        logs.append(f"  - 이번 실행 대상 티커: {len(pending_tickers)}개")
+
+        if not pending_tickers:
+            progress(1.0, desc="완료!")
+            logs.append("\n✅ 이미 수집된 티커입니다. 추가 수집할 대상이 없습니다.")
+            return "\n".join(logs)
+
+        # ========== 3단계: 야후 파이낸스 데이터 수집 ==========
+        logs.append(f"\n📥 [3단계] 야후 파이낸스 데이터 수집 시작 (총 {len(pending_tickers)}개 티커)")
         logs.append(f"  - 배치 크기: {batch_size}")
         logs.append(f"  - 조회 기간(period): {period}")
         logs.append(f"  - 체크포인트 업로드 간격: {checkpoint_batch_size}개 티커")
@@ -553,10 +597,10 @@ def run_pipeline(
         success_count = 0
         fail_count = 0
         last_checkpoint_success_index = 0
-        total = len(all_tickers)
+        total = len(pending_tickers)
 
         def _upload_checkpoint(end_index):
-            nonlocal last_checkpoint_success_index
+            nonlocal last_checkpoint_success_index, base_all_df, base_recent_df
 
             if success_count <= last_checkpoint_success_index:
                 return
@@ -566,20 +610,31 @@ def run_pipeline(
                 f"(누적 성공 {success_count}개)"
             )
 
+            if not all_data_frames:
+                return
+
             checkpoint_all_df = pd.concat(all_data_frames, ignore_index=True)
             checkpoint_recent_df = pd.concat(recent_30d_frames, ignore_index=True)
 
-            result_all_ckpt = upload_dataset_to_hf(checkpoint_all_df, all_dataset_name, hf_token)
-            result_30d_ckpt = upload_dataset_to_hf(checkpoint_recent_df, recent_dataset_name, hf_token)
+            base_all_df = _merge_rows(base_all_df, checkpoint_all_df)
+            base_recent_df = _merge_rows(base_recent_df, checkpoint_recent_df)
+            base_recent_df = filter_last_30_days(base_recent_df)
+
+            result_all_ckpt = upload_dataset_to_hf(base_all_df, all_dataset_name, hf_token)
+            result_30d_ckpt = upload_dataset_to_hf(base_recent_df, recent_dataset_name, hf_token)
 
             logs.append(f"  - all 체크포인트: {result_all_ckpt}")
             logs.append(f"  - 30d 체크포인트: {result_30d_ckpt}")
 
+            all_data_frames.clear()
+            recent_30d_frames.clear()
+            gc.collect()
+
             last_checkpoint_success_index = success_count
 
-        for i, ticker in enumerate(all_tickers):
+        for i, ticker in enumerate(pending_tickers):
             # 진행률 업데이트
-            progress_pct = (i + 1) / total
+            progress_pct = 0.1 + ((i + 1) / total) * 0.75
             progress(progress_pct, desc=f"수집 중: {ticker} ({i + 1}/{total})")
 
             ticker_df = fetch_ticker_data(ticker, period=period)
@@ -606,41 +661,37 @@ def run_pipeline(
 
         logs.append(f"\n📊 수집 완료: 성공 {success_count}개 / 실패 {fail_count}개")
 
-        if not all_data_frames:
+        if success_count == 0:
             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 = pd.concat(recent_30d_frames, ignore_index=True)
-        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()}")
+        # ========== 4단계: 마지막 미반영 체크포인트 반영 ==========
+        progress(0.9, desc="마지막 체크포인트 반영 중...")
+        logs.append("\n🔧 [4단계] 마지막 미반영 데이터 반영 중...")
 
         if checkpoint_batch_size > 0 and success_count > last_checkpoint_success_index:
             logs.append("\n💾 [체크포인트] 마지막 미반영 구간 업로드")
             _upload_checkpoint(total)
 
+        if checkpoint_batch_size == 0 and all_data_frames:
+            logs.append("\n💾 [최종반영] 중간 업로드 없이 누적된 데이터 반영")
+            _upload_checkpoint(total)
+
+        logs.append(f"  - 전체 데이터: {len(base_all_df)}행 x {len(base_all_df.columns)}열")
+        logs.append(f"  - 고유 티커 수: {base_all_df['Ticker'].nunique()}")
+        logs.append(f"  - 30일 데이터: {len(base_recent_df)}행 x {len(base_recent_df.columns)}열")
+        logs.append(f"  - 30일 고유 티커 수: {base_recent_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)
+        result_all = upload_dataset_to_hf(base_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)
+        result_30d = upload_dataset_to_hf(base_recent_df, recent_dataset_name, hf_token)
         logs.append(f"  {result_30d}")
 
         # ========== 완료 ==========