코스피 시황 자동화 파이프라인 — Python·n8n·Ollama로 비용 0원 만들기

-- 일별 지수 데이터 (파이프라인의 핵심 원본)
CREATE TABLE IF NOT EXISTS market_daily (
    trade_date      DATE     NOT NULL UNIQUE,  -- 거래일 (중복 방지)
    open_price      REAL,                      -- 시가
    high_price      REAL,                      -- 고가
    low_price       REAL,                      -- 저가
    close_price     REAL     NOT NULL,         -- 종가 (필수)
    change_rate     REAL,                      -- 등락률 % (+상승/-하락)
    foreign_net     BIGINT   DEFAULT 0,        -- 외국인 순매수 (억원)
    created_at      DATETIME DEFAULT CURRENT_TIMESTAMP
);

-- AI 분석 결과 (심리 점수가 핵심)
CREATE TABLE IF NOT EXISTS ai_analysis (
    trade_date      DATE     NOT NULL UNIQUE,
    summary_3line   TEXT,                      -- 핵심 3줄 요약
    sentiment_score INTEGER                    -- 투자 심리 1~100
                    CHECK(sentiment_score BETWEEN 1 AND 100),
    keywords        TEXT,                      -- JSON 배열
    market_context  TEXT,                      -- 시황 인과관계 해석
    tomorrow_watch  TEXT                       -- 내일 관전 포인트
);

-- 30일 통합 뷰 (분석용 — 조인 없이 한 번에)
CREATE VIEW IF NOT EXISTS v_daily_summary AS
SELECT m.trade_date, m.close_price, m.change_rate,
       a.sentiment_score, a.keywords, b.wp_status
FROM market_daily m
LEFT JOIN ai_analysis a ON m.trade_date = a.trade_date
LEFT JOIN blog_posts  b ON m.trade_date = b.trade_date
ORDER BY m.trade_date DESC LIMIT 30;

# SQLite DB 생성 + 스키마 적용
sqlite3 kospi.db < schema.sql

# 제대로 만들어졌는지 확인
sqlite3 kospi.db ".tables"
# 출력: ai_analysis  blog_posts  market_daily  news_raw  pipeline_log

import FinanceDataReader as fdr

# 1순위: FinanceDataReader (KRX 공식 데이터)
df = fdr.DataReader("KOSPI", trade_date, trade_date)

# 자동 fallback: FDR 실패 시 Yahoo Finance로 전환
# ^KS11 = KOSPI 티커 (Yahoo Finance 코드)
import yfinance as yf
ticker = yf.Ticker("^KS11")
df = ticker.history(start=trade_date, end=trade_date)

from bs4 import BeautifulSoup

url  = "https://finance.naver.com/news/news_list.naver?mode=HEAD"
resp = requests.get(url, headers=headers, timeout=10)
soup = BeautifulSoup(resp.text, "html.parser")

# 헤드라인 최대 15개 수집
items = soup.select(".realtimeNewsList .articleSubject a")[:15]
for item in items:
    title = item.get_text(strip=True)
    db.execute("INSERT OR IGNORE INTO news_raw ...")
    # OR IGNORE: 같은 날 중복 제목 자동 방지

prompt = f"""당신은 한국 주식 시장 전문 애널리스트입니다.
{trade_date} 코스피 데이터를 분석하고 JSON으로만 답변하세요.

## 코스피 데이터
- 종가: {close_price:,.2f}p ({change_rate:+.2f}%)
## 오늘 주요 뉴스
{news_text}

## 요청 형식 (JSON만, 다른 텍스트 금지)
{{
  "summary_3line": "요약1\\n요약2\\n요약3",
  "sentiment_score": 1~100 정수,
  "keywords": ["키워드1", ..., "키워드5"],
  "market_context": "인과관계 3~4문장",
  "tomorrow_watch": "내일 관전 포인트",
  "blog_title_candidates": ["제목1", "제목2", "제목3"]
}}"""

payload = {
    "model":  "qwen2.5:14b",
    "prompt": prompt,
    "stream": False,
    "options": {"temperature": 0.3}  # 낮을수록 일관된 분석
}

resp = requests.post(f"{OLLAMA_URL}/api/generate",
                     json=payload, timeout=180)

# 마크다운 코드블록 제거 후 JSON 파싱
clean    = re.sub(r"```(?:json)?|```", "", result_text).strip()
analysis = json.loads(clean)

# 전체 실행 (기본)
python kospi_pipeline.py

# 특정 날짜 지정
python kospi_pipeline.py --date 2026-05-29

# 수집만 (뉴스·지수)
python kospi_pipeline.py --step collect

# AI 분석만 (수집은 이미 됐을 때)
python kospi_pipeline.py --step analyze

# 블로그 초안만 재생성
python kospi_pipeline.py --step post

1. http://192.168.1.253:5678 접속
2. 왼쪽 메뉴 → Workflows → New Workflow
3. 우측 상단 ⋮ 메뉴 → Import from File
4. n8n_kospi_workflow.json 선택
5. 각 노드 클릭 → 크리덴셜 선택 (SQLite / WordPress / Telegram)
6. 우측 상단 토글 → Active ON
7. 내일 15:40까지 기다리거나 → Test workflow로 즉시 테스트

# 필수 패키지
pip install FinanceDataReader requests beautifulsoup4 python-dotenv

# fallback용 (FinanceDataReader 실패 시)
pip install yfinance

# 설치 확인
python -c "import FinanceDataReader; print('FDR OK')"

cp .env.example .env
nano .env

# 반드시 수정할 항목:
DB_PATH=./kospi.db
OLLAMA_URL=http://192.168.1.253:11434  # 본인 Ollama 주소
OLLAMA_MODEL=qwen2.5:14b

# WordPress (선택, 설정하면 자동 발행)
WP_URL=https://agibop.com
WP_USER=admin
WP_APP_PASSWORD=xxxx xxxx xxxx xxxx xxxx xxxx

# 1. DB 초기화
sqlite3 kospi.db < schema.sql

# 2. 최근 거래일로 테스트 실행
python kospi_pipeline.py --date 2026-05-29

# 정상 출력 예시:
# [15:41:02] [INFO] [collect_market] ✓ 수집 완료 | 종가: 2,687.25 | 등락: +1.24%
# [15:41:05] [INFO] [collect_news] ✓ 뉴스 12건 수집 완료
# [15:42:18] [INFO] [ollama_analyze] ✓ 분석 완료 | 심리점수: 72 | 소요: 73.2s
# [15:42:19] [INFO] [generate_post] ✓ 블로그 초안 생성 완료
# [15:42:19] [INFO] [main] ✅ 전체 파이프라인 완료 | 총 소요: 77.4초

# 3. DB 결과 확인
sqlite3 kospi.db "SELECT trade_date, close_price, change_rate, sentiment_score FROM v_daily_summary LIMIT 5;"

# 4. 생성된 블로그 HTML 확인
sqlite3 kospi.db "SELECT title, substr(content_html, 1, 200) FROM blog_posts ORDER BY created_at DESC LIMIT 1;"

-- 1. 심리점수 70 이상인 날의 다음날 평균 수익률 (백테스팅)
SELECT
    AVG(m2.change_rate) AS avg_next_day_return,
    COUNT(*)            AS sample_count
FROM ai_analysis a1
JOIN market_daily m1 ON a1.trade_date = m1.trade_date
JOIN market_daily m2 ON m2.trade_date = (
    SELECT trade_date FROM market_daily
    WHERE trade_date > a1.trade_date ORDER BY trade_date LIMIT 1
)
WHERE a1.sentiment_score >= 70;

-- 2. 반도체 키워드 등장일 평균 등락률
SELECT
    AVG(m.change_rate) AS avg_return,
    COUNT(*)           AS days_count
FROM market_daily m
JOIN ai_analysis a ON m.trade_date = a.trade_date
WHERE a.keywords LIKE '%반도체%';

-- 3. 요일별 평균 코스피 등락률
SELECT
    CASE strftime('%w', trade_date)
        WHEN '1' THEN '월요일'
        WHEN '2' THEN '화요일'
        WHEN '3' THEN '수요일'
        WHEN '4' THEN '목요일'
        WHEN '5' THEN '금요일'
    END AS weekday,
    ROUND(AVG(change_rate), 3) AS avg_return,
    COUNT(*)                   AS count
FROM market_daily
GROUP BY strftime('%w', trade_date)
ORDER BY strftime('%w', trade_date);

-- 4. 월별 심리점수 평균 추이
SELECT
    strftime('%Y-%m', trade_date) AS month,
    ROUND(AVG(sentiment_score), 1) AS avg_sentiment,
    ROUND(AVG(close_price), 2)     AS avg_close
FROM v_daily_summary
GROUP BY strftime('%Y-%m', trade_date)
ORDER BY month;