파이프라인 변환 베이스라인과 문서 추가

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)를 참고하세요.

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

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.
+

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.
+

pipeline/transforms/__init__.py

@@ -0,0 +1,2 @@
+"""Data transformation helpers for FinBridge pipelines."""
+

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
+