목차 — M0 데이터 수집 모듈
1. 모듈 개요
M0 데이터 수집 모듈은 Alchemetric 플랫폼의 기반 모듈로, 13개 데이터 소스에서 7개 API를 통해 주가/재무/거시경제/뉴스/공시/수급/대체 데이터를 수집하고, 8개 DB 테이블에 적재하는 파이프라인이다.
| 항목 | 수치 |
|---|---|
| 데이터 소스 | 13개 (yfinance, FDR, KIS API, DART API, FRED API, Naver News API, KR-FinBert-SC, Google Trends, Naver 종토방) |
| API 키 필요 | 5개 (KIS, DART, FRED, Naver) — 미설정 시 Graceful Skip |
| DB 테이블 | 9개 (stock_price, stock_info, stock_fundamental, financial_statement, macro_indicator, news_sentiment, dart_disclosure, krx_supply_demand, alternative_data) |
| 코드 규모 | 29개 파일, 약 4,900줄 |
| 수집 Phase | 6단계 (가격→재무→거시→뉴스→공시/수급→대체데이터) |
2. 아키텍처 — 데이터 파이프라인
수집 파이프라인은 Scheduler → Service → Fetcher → Repository → DB 4계층 구조를 따른다. 각 계층은 단일 책임 원칙을 지키며, API 장애 시에도 다른 소스의 수집이 중단되지 않는 Graceful Degradation 전략을 적용했다.
Cron 기반 자동 실행 · DB 잡 설정 · Admin CRUD
| StockSvc | FundSvc | MacroSvc | NewsSvc | DiscSvc | AltSvc |
|
KR/US Fetcher |
KIS DART |
yfinance FDR · FRED |
Naver FinBert |
DART KIS(KRX) |
Trends Community |
| StockRepo | FundRepo | MacroRepo | NewsRepo | DiscRepo | AltRepo |
계층별 역할
| 계층 | 역할 | 위치 |
|---|---|---|
| Scheduler | Cron 기반 자동 트리거, DB에서 잡 설정 로드, Admin CRUD | scheduler/scheduler.py |
| Service | 비즈니스 로직 오케스트레이션, Fetcher 호출 → Repository 저장 | services/*.py |
| Fetcher | 외부 API 호출, 데이터 정규화, Retry/Rate Limit 처리 | data_collector/*.py |
| Repository | DB 추상화, Upsert(Insert or Update), PostgreSQL/SQLite 이중 지원 | repositories/*.py |
3. 데이터 소스 (13개, 7 API)
| # | 소스 | 수집 대상 | 인증 | Phase |
|---|---|---|---|---|
| 1 | FinanceDataReader | 한국 주가 OHLCV | - | Phase 1 |
| 2 | yfinance | 미국 주가 OHLCV | - | Phase 1 |
| 3 | KIS API | PER, PBR, EPS, 시총, 외국인비율 | APP_KEY |
Phase 2 |
| 4 | DART API | 매출, 영업이익, ROE, 부채비율 | API_KEY |
Phase 2 |
| 5 | yfinance (거시) | 환율, VIX, KOSPI, S&P500, WTI, Gold, US 10Y | - | Phase 3 |
| 6 | FDR (거시) | 한국 3년물 국채금리 | - | Phase 3 |
| 7 | FRED API | 기준금리(DFF), 달러인덱스(DTWEXBGS), CPI(CPIAUCSL) | API_KEY |
Phase 3 |
| 8 | Naver News API | 종목별/시장 뉴스 기사 | CLIENT_ID |
Phase 4 |
| 9 | KR-FinBert-SC | 감성분석 (positive/negative/neutral) | - (로컬) | Phase 4 |
| 10 | DART API (공시) | 공시 목록, 유형 분류, 감성 점수 | API_KEY |
Phase 5 |
| 11 | KIS API (수급) | 공매도 비율, 프로그램 매수/매도량 | APP_KEY |
Phase 5 |
| 12 | Google Trends | 종목별 검색 트렌드 (주간→일간 보간) | - | Phase 7 |
| 13 | Naver 종토방 | 종목별 게시글/댓글 수 (HTML 크롤링) | - | Phase 7 |
Graceful Degradation: API 키가 없는 소스는 에러 없이 건너뛴다. 5개 API 키 중 하나도 없어도 Phase 1(주가) 수집은 정상 동작한다.
4. Phase별 수집 상세
Phase 1 — 주가 데이터 (OHLCV)
| 시장 | 라이브러리 | 병렬화 | Retry |
|---|---|---|---|
| 한국 (KOSPI/KOSDAQ) | FinanceDataReader | ThreadPoolExecutor (8 workers) | 3회, 1s, backoff 2x |
| 미국 (NYSE/NASDAQ/S&P500) | yfinance | 내장 배치 다운로드 | 3회, 2s, backoff 2x |
# DataPipeline — 수집 진입점 pipeline = DataPipeline() result = pipeline.fetch( market="KOSPI", start_date="2024-01-01", max_workers=8, # 한국: ThreadPool 병렬 ) # result.data = {code: DataFrame, ...} # result.success_count, result.failed_count # StockService — DB 적재 svc = StockService(pipeline=pipeline) saved = svc.save_to_db(result.data, "KOSPI") # → StockRepository.upsert_prices() → stock_price 테이블
Phase 2 — 펀더멘털 (KIS + DART)
| 소스 | 수집 데이터 | 주기 | 기술 포인트 |
|---|---|---|---|
| KIS API | PER, PBR, EPS, BPS, 시총, 배당률, 외국인비율, 기관/외인/개인 순매수 | 일간 | OAuth2 토큰 캐싱 (23h), Rate Limit 0.1s |
| DART API | 매출, 영업이익, 순이익, ROE, ROA, 부채비율 | 분기 | 연결(CFS) 우선, 키워드 기반 항목 추출 |
# KIS OAuth2 토큰 캐싱 def _get_access_token(self): now = datetime.now() if self._access_token and now < self._token_expires_at: return self._access_token # 캐시 재사용 # 새 토큰 발급 (24h 만료 → 23h로 안전 마진) resp = requests.post(token_url, json=body) self._access_token = resp["access_token"] self._token_expires_at = now + timedelta(hours=23)
Phase 3 — 거시경제 지표 (11개)
# MacroFetcher — 3개 소스에서 11개 지표 수집 class MacroFetcher: def fetch_all(self, start_date, end_date): results = [] # ① yfinance — 7개 (개별 다운로드, SQLite lock 회피) for symbol in ["USDKRW=X", "^VIX", "^KS11", "^GSPC", "CL=F", "GC=F", "^TNX"]: data = yf.download(symbol, threads=False) results.extend(self._to_records(data)) # ② FDR — 1개 (한국 3년물 국채) data = fdr.DataReader("KR3YT=RR", start_date) results.extend(self._to_records(data)) # ③ FRED — 3개 (기준금리, 달러인덱스, CPI) if self.fred_available: for series_id in ["DFF", "DTWEXBGS", "CPIAUCSL"]: data = self.fred.get_series(series_id) results.extend(self._to_records(data)) return MacroFetchResult(records=results)
Phase 4 — 뉴스 센티먼트 (Naver + KR-FinBert-SC)
# 수집 → 분석 → 저장 파이프라인 NewsFetcher.fetch_stock_news(code, stock_name) # → Naver News API (max 1000건, 0.1s rate limit) # → HTML 태그 제거, 날짜 파싱 │ ▼ SentimentAnalyzer.analyze(texts, batch_size=32) # → KR-FinBert-SC (Lazy Singleton) # → score: -1.0 ~ +1.0, label: pos/neg/neutral │ ▼ NewsRepository.upsert_articles(records) # → news_sentiment 테이블
Phase 5 — 공시 / 수급
| 구분 | 소스 | 데이터 | 기술 포인트 |
|---|---|---|---|
| 5A 공시 | DART API | 공시 목록, 7개 유형 분류 (실적 1.0 ~ 기타 0.2) | KR-FinBert-SC 감성분석 연동 |
| 5B 수급 | KIS API (TR) | 공매도 비율, 프로그램 매수/매도량 | 56일/45일 청크, Rate Limit 0.12s |
Phase 7 — 대체 데이터 (Google Trends + Naver 종토방)
| 구분 | 소스 | 데이터 | 기술 포인트 |
|---|---|---|---|
| 7A 트렌드 | Google Trends (pytrends) | 종목별 검색량 (0~100), 일간 보간값 | Retry 3회, 2s Rate Limit, 주간→일간 선형 보간 |
| 7B 커뮤니티 | Naver 종토방 (BeautifulSoup) | 일별 게시글 수, 댓글 수 | HTML 파싱, 0.4s Rate Limit, 최대 20페이지 |
# GoogleTrendsFetcher — 주간 데이터를 일간으로 보간 fetcher = GoogleTrendsFetcher() result = fetcher.fetch_stock_trend( code="005930", stock_name="삼성전자", market="KOSPI", days=90, ) # → google_trend_value (주간 0~100) # → google_trend_interpolated (일간 보간) # NaverCommunityFetcher — 종토방 크롤링 fetcher = NaverCommunityFetcher() result = fetcher.fetch_community_data( code="005930", market="KOSPI", days=30, max_pages=20, ) # → community_post_count (일별 게시글) # → community_comment_count (일별 댓글)
5. DB 스키마 (9 테이블)
| 테이블 | Phase | Unique Key | 주요 컬럼 |
|---|---|---|---|
| stock_price | 1 | (market, code, date) | OHLCV |
| stock_info | 1 | (market, code) | 종목명, 섹터, 산업 |
| stock_fundamental | 2 | (market, code, date) | PER, PBR, EPS, 시총, 외인비율 |
| financial_statement | 2 | (market, code, period) | 매출, 영업이익, ROE, 부채비율 |
| macro_indicator | 3 | (date, indicator_name) | EAV — 11개 지표, value, source |
| news_sentiment | 4 | (url, code) | 제목, 감성점수(-1~+1), 라벨 |
| dart_disclosure | 5A | (rcept_no, code) | 보고서명, 유형(7분류), 감성점수 |
| krx_supply_demand | 5B | (market, code, date) | 공매도비율, 프로그램매수/매도량 |
| alternative_data | 7 | (market, code, date) | 검색트렌드(주간/일간), 게시글수, 댓글수 |
Upsert 전략: 모든 테이블은 Unique Constraint 기반 Upsert(Insert or Update)를 사용한다. PostgreSQL은
on_conflict_do_update(constraint=...), SQLite는on_conflict_do_update(index_elements=...)로 DB별 분기 처리. 동일 데이터를 재수집해도 안전하게 갱신된다 (멱등성).
6. 엔지니어링 패턴
6.1 Retry 데코레이터
모든 외부 API 호출에 @retry 데코레이터를 적용한다. 지수 백오프(exponential backoff)로 일시적 장애를 자동 복구한다.
@retry(max_attempts=3, delay=1.0, backoff=2.0, module="kis") def fetch_stock_info(self, code): # 1차 실패 → 1s 대기 → 2차 실패 → 2s 대기 → 3차 시도 ...
6.2 Rate Limiting
| 소스 | 제한 | 구현 |
|---|---|---|
| KIS API | 20 req/sec | time.sleep(0.1) |
| KRX (KIS TR) | 청크별 | time.sleep(0.12) |
| Naver News | 페이지별 | time.sleep(0.1) |
| Google Trends | 요청별 | time.sleep(2.0) |
| Naver 종토방 | 페이지별 | time.sleep(0.4) |
| yfinance | - | 내장 배치 처리 |
6.3 Graceful Degradation
API 키가 없거나 외부 서비스 장애 시에도 수집 가능한 데이터는 정상 처리된다.
# FRED API 키 없음 → 8/11 거시지표만 수집 (나머지 3개 skip) # KIS API 키 없음 → 펀더멘털/수급 skip, 주가는 정상 수집 # Naver 키 없음 → 뉴스 skip, 나머지 전부 정상 # 개별 종목 실패 → 해당 종목만 skip, 나머지 계속 처리 # (단, 50% 이상 실패 시 critical 경고)
6.4 Singleton 패턴
| 클래스 | 이유 |
|---|---|
JobScheduler |
APScheduler 인스턴스 중복 생성 방지 |
SentimentAnalyzer |
BERT 모델 로딩 1회 (약 3~5초), 이후 재사용 |
6.5 DB Upsert (이중 지원)
# PostgreSQL stmt = pg_insert(Model).values(records) stmt = stmt.on_conflict_do_update( constraint="uq_market_code_date", set_={col: stmt.excluded[col] for col in update_fields} ) # SQLite stmt = sqlite_insert(Model).values(records) stmt = stmt.on_conflict_do_update( index_elements=["market", "code", "date"], set_={col: stmt.excluded[col] for col in update_fields} )
7. 코드 구조 & 통계
| 계층 | 파일 | 라인 | 비중 |
|---|---|---|---|
| Scheduler | scheduler/scheduler.py | 803 | 18% |
| Pipeline | pipeline.py | 321 | 7% |
| Fetcher (10개) | kr/us/kis/dart/macro/news/disclosure/krx/trends/community | 1,771 | 33% |
| Service (6개) | stock/fundamental/macro/news/disclosure/alternative_service | 685 | 13% |
| Repository (6개) | stock/fundamental/macro/news/disclosure/alternative_repo | 796 | 16% |
| Model (6개) | stock/fundamental/macro/news/disclosure/alternative | 375 | 7% |
| Config | config.py | 128 | 3% |
합계: 29개 파일, 약 4,900줄. Fetcher 계층(33%)이 가장 크며, 이는 7개 외부 API의 각기 다른 인터페이스와 데이터 정규화 로직을 담고 있기 때문이다.
파일 트리
app/ ├── data_collector/ │ ├── pipeline.py # 수집 오케스트레이션 │ ├── kr_fetcher.py # 한국 주가 (FDR) │ ├── us_fetcher.py # 미국 주가 (yfinance) │ ├── kis_fetcher.py # KIS 펀더멘털 + OAuth2 │ ├── dart_fetcher.py # DART 재무제표 │ ├── macro_fetcher.py # 거시지표 (yf+FDR+FRED) │ ├── news_fetcher.py # Naver News API │ ├── sentiment_analyzer.py # KR-FinBert-SC (Singleton) │ ├── disclosure_fetcher.py # DART 공시 목록 │ ├── krx_fetcher.py # KRX 수급 (via KIS TR) │ ├── stock_codes.py # 종목 코드 관리 (lru_cache) │ ├── google_trends_fetcher.py # Google Trends (pytrends, 160줄) │ └── naver_community_fetcher.py # Naver 종토방 크롤링 (191줄) ├── scheduler/ │ └── scheduler.py # APScheduler 잡 관리 (JobScheduler) ├── services/ │ ├── stock_service.py │ ├── fundamental_service.py │ ├── macro_service.py │ ├── news_service.py │ ├── disclosure_service.py │ └── alternative_service.py ├── repositories/ │ ├── stock_repository.py │ ├── fundamental_repository.py │ ├── macro_repository.py │ ├── news_repository.py │ ├── disclosure_repository.py │ └── alternative_repository.py └── models/ ├── stock.py, fundamental.py, macro.py ├── news.py, disclosure.py, alternative.py └── config.py
8. 스케줄러 연동
모든 수집 작업은 JobScheduler를 통해 Cron 기반으로 자동 실행된다. DB에 저장된 잡 설정을 로드하며, Admin 대시보드에서 CRUD가 가능하다.
| Job Type | 기본 스케줄 | 호출 Service |
|---|---|---|
fundamental_collect |
18:00 KST | FundamentalService.collect_fundamentals() |
macro_collect |
18:30 KST | MacroService.collect() |
news_collect |
19:00 KST | NewsService.collect() |
disclosure_collect |
19:30 KST | DisclosureService.collect_disclosures() |
supply_collect |
19:30 KST | DisclosureService.collect_supply_demand() |
alternative_collect |
19:45 KST | AlternativeService.collect_all() |
ml_train |
20:00 KST | 8단계 파이프라인 (수집 → 피처 → 학습) |
# 스케줄러 실행 흐름 scheduler = JobScheduler.get_instance() scheduler.load_jobs_from_db() # ScheduleJob 테이블에서 잡 로드 scheduler.start() # APScheduler 백그라운드 실행 # Admin에서 즉시 실행 POST /scheduler/run/{job_id} # → scheduler.run_now(job_id) # 실행 결과 → ScheduleLog 테이블에 자동 기록 # status: running → success | partial | failed
핵심 설계 결정: API 키 미설정 시 해당 소스만 건너뛰고, 수집 가능한 데이터는 정상 적재한다. 50% 이상 실패 시 critical 경고를 로그에 남기되, 파이프라인 자체는 중단하지 않는다. 모든 DB 쓰기는 Upsert 기반이므로 동일 데이터를 재수집해도 안전하다.
'Project > 🚀 AI Finance 2026' 카테고리의 다른 글
| #7 [모듈1-1] 한국어 금융 뉴스의 감성 sentiment 분석 (0) | 2026.02.19 |
|---|---|
| #6 [모듈1] 데이터 분석 - 개발문서 (0) | 2026.02.19 |
| #4 Alchemetric API 정의서 (0) | 2026.02.06 |
| #3 Alchemetric ERD (0) | 2026.02.03 |
| #2 Alchemetric 시스템 아키텍쳐 (0) | 2026.02.03 |