diff --git a/README.md b/README.md index fe3fc05..7f76d9c 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,11 @@ | 테스트 | pytest, httpx | | 인프라 | Docker Compose | +## 개발 문서 + +- [Data Pipeline Design](docs/data_pipeline_design.md) +- [Data Contract](docs/data_contract.md) + ## 브랜치 전략 GitHub Flow 기반으로 운영합니다. 자세한 규칙은 [.github/BRANCH_STRATEGY.md](.github/BRANCH_STRATEGY.md)를 참고하세요. diff --git a/data/seed/news_seed.csv b/data/seed/news_seed.csv new file mode 100644 index 0000000..d06635b --- /dev/null +++ b/data/seed/news_seed.csv @@ -0,0 +1,7 @@ +symbol,title,body,url,source,published_at +005930.KS,삼성전자 HBM 공급 기대감으로 반도체 실적 개선 전망,AI 서버 수요와 메모리 가격 상승이 영업이익 개선을 견인할 가능성이 높다는 분석이다.,https://example.com/finbridge/news/005930-20260615,FinBridge Seed,2026-06-15T09:00:00+09:00 +000660.KS,SK하이닉스 차세대 HBM 수주 확대 기대,글로벌 AI 반도체 투자 확대와 공급 계약 논의가 주가 강세 요인으로 거론됐다.,https://example.com/finbridge/news/000660-20260615,FinBridge Seed,2026-06-15T10:10:00+09:00 +005380.KS,현대차 전기차 수요 둔화 우려에도 자율주행 협력 확대,전기차 판매 성장 둔화 리스크가 있으나 자율주행 기술 협력은 중장기 성장 요인이다.,https://example.com/finbridge/news/005380-20260614,FinBridge Seed,2026-06-14T14:30:00+09:00 +035420.KS,NAVER AI 검색 고도화와 광고 매출 성장 기대,AI 기반 검색 품질 개선과 커머스 광고 매출 증가가 실적 개선으로 이어질 수 있다.,https://example.com/finbridge/news/035420-20260613,FinBridge Seed,2026-06-13T08:45:00+09:00 +TSLA,Tesla faces EV demand risk while FSD partnership talks continue,Analysts noted weak EV demand risk but highlighted autonomous driving licensing as a possible growth driver.,https://example.com/finbridge/news/tsla-20260612,FinBridge Seed,2026-06-12T12:00:00+00:00 +NVDA,NVIDIA revenue growth remains strong on AI GPU demand,AI infrastructure spending and GPU supply constraints supported positive earnings expectations.,https://example.com/finbridge/news/nvda-20260612,FinBridge Seed,2026-06-12T12:30:00+00:00 diff --git a/docs/data_contract.md b/docs/data_contract.md new file mode 100644 index 0000000..226f9fb --- /dev/null +++ b/docs/data_contract.md @@ -0,0 +1,61 @@ +# FinBridge Data Contract + +## OHLCV Input + +`pipeline.collectors.dtos.stock.OhlcvRecord` + +| Field | Type | Rule | +|---|---|---| +| `symbol` | `str` | Ticker symbol, normalized to uppercase in transforms | +| `date` | `date` | Candle date | +| `open` | `float` | `>= 0` | +| `high` | `float` | `>= 0` | +| `low` | `float` | `>= 0` | +| `close` | `float` | `>= 0` | +| `volume` | `int` | `>= 0` | + +## Finance Metrics Output + +`pipeline.transforms.finance_metrics.FinanceMetricRecord` + +| Field | Calculation | +|---|---| +| `daily_return_pct` | `(close - previous_close) / previous_close * 100` | +| `intraday_volatility_pct` | `(high - low) / close * 100` | +| `volume_change_pct` | `(volume - previous_volume) / previous_volume * 100` | +| `ma_5` | 5-candle close moving average | +| `ma_20` | 20-candle close moving average | +| `signal_label` | Research label, not investment advice | + +Signal labels: + +| Condition | Label | +|---|---| +| First row per symbol | `baseline` | +| Return `>= 3%` and volume change `>= 20%` | `bullish_breakout` | +| Return `<= -3%` and volume change `>= 20%` | `bearish_breakdown` | +| Absolute return `>= 2%` | `price_move` | +| Otherwise | `normal` | + +## News Baseline Input + +`pipeline.transforms.nlp_rules.NewsInput` + +| Field | Required | Rule | +|---|---|---| +| `symbol` | Yes | Related ticker | +| `title` | Yes | News title | +| `body` | No | News body or summary | +| `url` | No | Deduplication candidate | +| `source` | No | Provider/source name | +| `published_at` | No | Source timestamp | + +## News Baseline Output + +| Object | Purpose | +|---|---| +| `SentimentResult` | Rule-based sentiment score, label, and topics | +| `EventCandidate` | News-derived event candidate for later graph loading | + +The rule-based NLP module is a deterministic MVP baseline. It should be replaced or supplemented by model-based sentiment and extraction after the News API collector is ready. + diff --git a/docs/data_pipeline_design.md b/docs/data_pipeline_design.md new file mode 100644 index 0000000..6e19d89 --- /dev/null +++ b/docs/data_pipeline_design.md @@ -0,0 +1,69 @@ +# FinBridge Data Pipeline Design + +## Goal + +FinBridge collects market data, derives research features, and prepares context for SQL, Vector RAG, and GraphRAG workflows. The current repository already has a yfinance OHLCV collector and a PostgreSQL `candles` repository, so the next useful layer is transformation and validation rather than another collector. + +## Current Baseline + +```text +yfinance + -> pipeline.collectors.stock_data_collector + -> OhlcvRecord + -> candles table / CandleRepository +``` + +The MVP gap after collection is: + +- price feature generation +- news sentiment/topic/event baseline +- quality checks after persistence +- Neo4j graph loading from structured context +- orchestration through CLI/Airflow + +## Recommended Flow + +```mermaid +flowchart LR + yfinance["yfinance OHLCV"] + collector["collect_ohlcv / collect_ohlcvs"] + candles["PostgreSQL candles"] + metrics["finance metrics transform"] + metricStore["price feature table"] + news["news seed/API"] + nlp["sentiment/topic/event rules"] + graph["Neo4j GraphRAG context"] + qa["pipeline_runs / quality_checks"] + + yfinance --> collector --> candles --> metrics --> metricStore + news --> nlp --> graph + metricStore --> qa +``` + +## Layering + +| Layer | Repository location | Purpose | +|---|---|---| +| Collect | `pipeline/collectors/` | External source ingestion | +| Transform | `pipeline/transforms/` | Pure, testable feature/event generation | +| Persist | `app/repositories/` | Async SQLAlchemy and Neo4j persistence | +| Serve | `app/apis/`, `app/services/` | API and RAG routing | +| Operate | `scripts/`, `dags/` | Repeatable jobs and validation | + +## Why Transforms First + +The repo already has a tested collector. Adding another yfinance loader would duplicate logic. Pure transform modules are low risk because they: + +- do not require network access +- do not require database access +- are easy to unit test +- can later be called from Kafka consumers, Spark jobs, or API services + +## Next Integration Steps + +1. Persist collected OHLCV into `candles`. +2. Use `pipeline.transforms.finance_metrics.build_finance_metrics()` to derive return, volatility, moving average, and research labels. +3. Add a price feature SQLAlchemy model and repository. +4. Use `pipeline.transforms.nlp_rules` as a deterministic baseline for news sentiment and topic extraction. +5. Add a Neo4j graph loader after structured event/topic tables exist. + diff --git a/pipeline/transforms/__init__.py b/pipeline/transforms/__init__.py new file mode 100644 index 0000000..c8a96c8 --- /dev/null +++ b/pipeline/transforms/__init__.py @@ -0,0 +1,2 @@ +"""Data transformation helpers for FinBridge pipelines.""" + diff --git a/pipeline/transforms/finance_metrics.py b/pipeline/transforms/finance_metrics.py new file mode 100644 index 0000000..7292d96 --- /dev/null +++ b/pipeline/transforms/finance_metrics.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +from collections import defaultdict +from dataclasses import dataclass +from datetime import date +from decimal import ROUND_HALF_UP, Decimal +from typing import Iterable + +from pipeline.collectors.dtos.stock import OhlcvRecord + +FOUR_PLACES = Decimal("0.0001") + + +@dataclass(frozen=True) +class FinanceMetricRecord: + symbol: str + date: date + close: Decimal + daily_return_pct: Decimal | None + intraday_volatility_pct: Decimal | None + volume_change_pct: Decimal | None + ma_5: Decimal | None + ma_20: Decimal | None + signal_label: str + + +def _decimal(value: float | int | Decimal) -> Decimal: + return Decimal(str(value)) + + +def _quantize(value: Decimal | None) -> Decimal | None: + if value is None: + return None + return value.quantize(FOUR_PLACES, rounding=ROUND_HALF_UP) + + +def pct_change(current: Decimal, previous: Decimal | None) -> Decimal | None: + if previous is None or previous == 0: + return None + return _quantize(((current - previous) / previous) * Decimal("100")) + + +def moving_average(values: list[Decimal], window: int) -> Decimal | None: + if len(values) < window: + return None + return _quantize(sum(values[-window:]) / Decimal(window)) + + +def classify_signal( + daily_return_pct: Decimal | None, + volume_change_pct: Decimal | None, +) -> str: + """Classify research candidates, not investment advice.""" + if daily_return_pct is None: + return "baseline" + volume_delta = volume_change_pct or Decimal("0") + if daily_return_pct >= Decimal("3") and volume_delta >= Decimal("20"): + return "bullish_breakout" + if daily_return_pct <= Decimal("-3") and volume_delta >= Decimal("20"): + return "bearish_breakdown" + if abs(daily_return_pct) >= Decimal("2"): + return "price_move" + return "normal" + + +def build_finance_metrics(records: Iterable[OhlcvRecord]) -> list[FinanceMetricRecord]: + by_symbol: dict[str, list[OhlcvRecord]] = defaultdict(list) + for record in records: + by_symbol[record.symbol.upper().strip()].append(record) + + metrics: list[FinanceMetricRecord] = [] + for symbol, symbol_records in by_symbol.items(): + closes: list[Decimal] = [] + previous_close: Decimal | None = None + previous_volume: Decimal | None = None + + for record in sorted(symbol_records, key=lambda item: item.date): + close = _decimal(record.close) + high = _decimal(record.high) + low = _decimal(record.low) + volume = _decimal(record.volume) + closes.append(close) + + daily_return = pct_change(close, previous_close) + volatility = _quantize(((high - low) / close) * Decimal("100")) if close else None + volume_change = pct_change(volume, previous_volume) + signal = classify_signal(daily_return, volume_change) + + metrics.append( + FinanceMetricRecord( + symbol=symbol, + date=record.date, + close=close, + daily_return_pct=daily_return, + intraday_volatility_pct=volatility, + volume_change_pct=volume_change, + ma_5=moving_average(closes, 5), + ma_20=moving_average(closes, 20), + signal_label=signal, + ) + ) + previous_close = close + previous_volume = volume + + return metrics + diff --git a/pipeline/transforms/nlp_rules.py b/pipeline/transforms/nlp_rules.py new file mode 100644 index 0000000..3a05aeb --- /dev/null +++ b/pipeline/transforms/nlp_rules.py @@ -0,0 +1,152 @@ +from __future__ import annotations + +import re +from dataclasses import dataclass +from datetime import date, datetime +from decimal import Decimal + +POSITIVE_WORDS = { + "상승", + "호조", + "개선", + "성장", + "증가", + "돌파", + "수주", + "흑자", + "강세", + "협력", + "positive", + "growth", + "beat", + "surge", + "record", +} + +NEGATIVE_WORDS = { + "하락", + "부진", + "감소", + "손실", + "적자", + "리스크", + "우려", + "규제", + "약세", + "negative", + "loss", + "miss", + "risk", + "probe", +} + +TOPIC_PATTERNS: dict[str, tuple[str, ...]] = { + "반도체": ("반도체", "hbm", "메모리", "foundry", "파운드리", "chip"), + "인공지능": ("ai", "인공지능", "llm", "gpu", "npu"), + "전기차": ("전기차", "ev", "battery", "배터리"), + "자율주행": ("자율주행", "autonomous", "fsd"), + "금리": ("금리", "fomc", "fed", "연준", "rate"), + "환율": ("환율", "달러", "원화", "exchange"), + "실적": ("실적", "매출", "영업이익", "earnings", "revenue", "profit"), + "규제": ("규제", "제재", "antitrust", "probe"), + "공급망": ("공급망", "supply chain", "수급"), +} + +EVENT_TYPE_PATTERNS: dict[str, tuple[str, ...]] = { + "earnings": ("실적", "매출", "영업이익", "earnings", "revenue", "profit"), + "policy": ("금리", "fomc", "fed", "연준", "규제", "제재"), + "partnership": ("협력", "계약", "공급", "수주", "partnership", "deal"), + "technology": ("hbm", "ai", "인공지능", "자율주행", "기술"), +} + + +@dataclass(frozen=True) +class NewsInput: + symbol: str + title: str + body: str | None = None + url: str | None = None + source: str | None = None + published_at: datetime | None = None + + +@dataclass(frozen=True) +class SentimentResult: + score: Decimal + label: str + topics: tuple[str, ...] + + +@dataclass(frozen=True) +class EventCandidate: + symbol: str + event_type: str + description: str + event_date: date + impact_label: str + topics: tuple[str, ...] + + +def _tokens(text: str) -> list[str]: + return re.findall(r"[0-9A-Za-z가-힣]+", text.lower()) + + +def extract_topics(text: str) -> tuple[str, ...]: + lowered = text.lower() + topics = [ + topic + for topic, patterns in TOPIC_PATTERNS.items() + if any(pattern.lower() in lowered for pattern in patterns) + ] + return tuple(dict.fromkeys(topics)) + + +def analyze_news(item: NewsInput) -> SentimentResult: + text = f"{item.title} {item.body or ''}".lower() + tokens = set(_tokens(text)) + positive = sum(1 for word in POSITIVE_WORDS if word.lower() in tokens or word.lower() in text) + negative = sum(1 for word in NEGATIVE_WORDS if word.lower() in tokens or word.lower() in text) + raw_score = positive - negative + + if raw_score > 0: + label = "positive" + elif raw_score < 0: + label = "negative" + else: + label = "neutral" + + score = Decimal(raw_score) / Decimal(max(positive + negative, 1)) + return SentimentResult( + score=score.quantize(Decimal("0.01")), + label=label, + topics=extract_topics(text), + ) + + +def event_from_news(item: NewsInput, sentiment: SentimentResult) -> EventCandidate | None: + text = f"{item.title} {item.body or ''}".lower() + event_type = next( + ( + candidate + for candidate, patterns in EVENT_TYPE_PATTERNS.items() + if any(pattern.lower() in text for pattern in patterns) + ), + None, + ) + + if event_type is None and sentiment.label == "neutral": + return None + if event_type is None: + event_type = "market_news" + + impact = sentiment.label if sentiment.label in {"positive", "negative"} else "neutral" + event_date = item.published_at.date() if item.published_at else date.today() + return EventCandidate( + symbol=item.symbol.upper().strip(), + event_type=event_type, + description=item.title, + event_date=event_date, + impact_label=impact, + topics=sentiment.topics, + ) + diff --git a/tests/unit/test_finance_metrics.py b/tests/unit/test_finance_metrics.py new file mode 100644 index 0000000..2383a19 --- /dev/null +++ b/tests/unit/test_finance_metrics.py @@ -0,0 +1,40 @@ +from datetime import date +from decimal import Decimal + +from pipeline.collectors.dtos.stock import OhlcvRecord +from pipeline.transforms.finance_metrics import build_finance_metrics, pct_change + + +def test_pct_change_returns_percent() -> None: + assert pct_change(Decimal("110"), Decimal("100")) == Decimal("10.0000") + + +def test_build_finance_metrics_classifies_breakout() -> None: + records = [ + OhlcvRecord( + symbol="aaa", + date=date(2026, 1, 1), + open=100, + high=101, + low=99, + close=100, + volume=100, + ), + OhlcvRecord( + symbol="aaa", + date=date(2026, 1, 2), + open=103, + high=106, + low=102, + close=105, + volume=130, + ), + ] + + metrics = build_finance_metrics(records) + + assert metrics[-1].symbol == "AAA" + assert metrics[-1].daily_return_pct == Decimal("5.0000") + assert metrics[-1].volume_change_pct == Decimal("30.0000") + assert metrics[-1].signal_label == "bullish_breakout" + diff --git a/tests/unit/test_nlp_rules.py b/tests/unit/test_nlp_rules.py new file mode 100644 index 0000000..ae47d5b --- /dev/null +++ b/tests/unit/test_nlp_rules.py @@ -0,0 +1,39 @@ +from datetime import datetime, timezone +from decimal import Decimal + +from pipeline.transforms.nlp_rules import NewsInput, analyze_news, event_from_news + + +def test_analyze_news_extracts_positive_topics() -> None: + item = NewsInput( + symbol="005930.KS", + title="삼성전자 HBM 수주와 AI 반도체 실적 개선", + body="영업이익 성장 기대가 커졌다.", + url="https://example.com/news", + source="test", + published_at=datetime(2026, 6, 1, tzinfo=timezone.utc), + ) + + sentiment = analyze_news(item) + + assert sentiment.label == "positive" + assert sentiment.score > Decimal("0") + assert "반도체" in sentiment.topics + assert "인공지능" in sentiment.topics + + +def test_event_from_news_builds_candidate() -> None: + item = NewsInput( + symbol="TSLA", + title="Tesla EV demand risk", + body="Weak demand risk remains.", + published_at=datetime(2026, 6, 1, tzinfo=timezone.utc), + ) + sentiment = analyze_news(item) + + event = event_from_news(item, sentiment) + + assert event is not None + assert event.symbol == "TSLA" + assert event.impact_label == "negative" +