#5 [모듈0] 데이터 수집

목차 — M0 데이터 수집 모듈

모듈 개요
아키텍처 — 데이터 파이프라인
데이터 소스 (13개, 7 API)
Phase별 수집 상세
DB 스키마 (9 테이블)
엔지니어링 패턴
코드 구조 & 통계
스케줄러 연동

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 전략을 적용했다.

DataScheduler (APScheduler)
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

▼ ▼ ▼ ▼ ▼ ▼

PostgreSQL / SQLite (9 tables)

계층별 역할

계층	역할	위치
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

내일의 내가 달라질 수 있다면

#5 [모듈0] 데이터 수집 - 개발문서

목차 — M0 데이터 수집 모듈

1. 모듈 개요

2. 아키텍처 — 데이터 파이프라인

계층별 역할

3. 데이터 소스 (13개, 7 API)

4. Phase별 수집 상세

Phase 1 — 주가 데이터 (OHLCV)

Phase 2 — 펀더멘털 (KIS + DART)

Phase 3 — 거시경제 지표 (11개)

Phase 4 — 뉴스 센티먼트 (Naver + KR-FinBert-SC)

Phase 5 — 공시 / 수급

Phase 7 — 대체 데이터 (Google Trends + Naver 종토방)

5. DB 스키마 (9 테이블)

6. 엔지니어링 패턴

6.1 Retry 데코레이터

6.2 Rate Limiting

6.3 Graceful Degradation

6.4 Singleton 패턴

6.5 DB Upsert (이중 지원)

7. 코드 구조 & 통계

파일 트리

8. 스케줄러 연동

'Project > 🚀 AI Finance 2026' 카테고리의 다른 글

티스토리툴바

#5 [모듈0] 데이터 수집 - 개발문서

목차 — M0 데이터 수집 모듈

1. 모듈 개요

2. 아키텍처 — 데이터 파이프라인

계층별 역할

3. 데이터 소스 (13개, 7 API)

4. Phase별 수집 상세

Phase 1 — 주가 데이터 (OHLCV)

Phase 2 — 펀더멘털 (KIS + DART)

Phase 3 — 거시경제 지표 (11개)

Phase 4 — 뉴스 센티먼트 (Naver + KR-FinBert-SC)

Phase 5 — 공시 / 수급

Phase 7 — 대체 데이터 (Google Trends + Naver 종토방)

5. DB 스키마 (9 테이블)

6. 엔지니어링 패턴

6.1 Retry 데코레이터

6.2 Rate Limiting

6.3 Graceful Degradation

6.4 Singleton 패턴

6.5 DB Upsert (이중 지원)

7. 코드 구조 & 통계

파일 트리

8. 스케줄러 연동

'Project > 🚀 AI Finance 2026' 카테고리의 다른 글

관련글

티스토리툴바