라즈베리파이 제로 e-ink 사용하기

라즈베리파이 + Waveshare e-Paper로 삼성전자/현대차 실시간 주가 모니터링

개요

데이터는 네이버 금융 API(m.stock.naver.com)에서 실시간으로 가져오고, e-ink 수명 보호를 위해 가격이 바뀔 때만 화면을 갱신하도록 구성했습니다. Waveshare 2.13” 모델 기준이고, 다른 사이즈는 코드 상단에서 모듈만 바꾸면 됩니다.​​​​​​​​​​​​​​​​

주요 특징

• 주요 특징
• 데이터 소스: 네이버 모바일 금융 API(m.stock.naver.com/api/stock/{code}/basic) — 별도 API 키 불필요, 응답 안정적

e-ink 보호 로직: • 가격/등락이 바뀌었을 때만 화면 갱신 (해시 비교) • 장중 1분, 장외 30분 주기로 자동 전환 • SIGTERM/SIGINT 수신 시 화면 클리어 후 sleep 모드 진입 화면 구성 (2.7” 264x176 가로 모드): • 상단: 현재 시각 + [장중/장외] 표시 • 종목별 영역에 종목명, 현재가(큰 글씨), ▲▼ 등락폭/등락률, 시가/고가/저가 확인이 필요한 부분

라이브러리 설치

1. 의존 패키지 설치

# 시스템 패키지
sudo apt update
sudo apt install -y python3-pip python3-pil python3-numpy fonts-nanum
sudo apt install -y python3-spidev python3-rpi.gpio

# 파이썬 패키지
pip3 install requests Pillow

# SPI 활성화 (raspi-config → Interface Options → SPI → Enable)
sudo raspi-config

2. Waveshare e-Paper 라이브러리 설치 코드 상단 libdir 경로가 위와 일치하는지 확인. 다른 모델 사용 시 from waveshare_epd import epd2in7 as epd_module 부분만 변경: • 2.13”: epd2in13_V3 • 2.9”: epd2in9_V2 • 4.2”: epd4in2 • 7.5”: epd7in5_V2

cd ~
git clone https://github.com/waveshare/e-Paper.git

3. 소스 코드 stock_monitor.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
삼성전자(005930) / 현대차(005380) 실시간 주가 모니터 + 주봉 차트
Raspberry Pi + Waveshare 2.13" e-Paper (250x122)

페이지 자동 전환:
  Page 0 - 요약 (양 종목)
  Page 1 - 삼성전자 + 주봉 차트
  Page 2 - 현대차 + 주봉 차트

Author: 치치 (dbaworks)
"""

import os
import re
import sys
import time
import logging
import signal
import requests
from datetime import datetime, timedelta, time as dtime
from PIL import Image, ImageDraw, ImageFont

libdir = os.path.expanduser('~/e-Paper/RaspberryPi_JetsonNano/python/lib')
if os.path.exists(libdir):
    sys.path.append(libdir)

from waveshare_epd import epd2in13_V4 as epd_module   # ← 모델에 맞게 변경

# ─────────────────────────────────────────────────────────────
# 설정
# ─────────────────────────────────────────────────────────────
STOCKS = [
    {'code': '005930', 'name': '삼성전자'},
    {'code': '005380', 'name': '현대차'},
]

CHART_WEEKS = 26                  # 주봉 표시 기간
DATA_REFRESH_OPEN  = 60           # 장중 시세 갱신 주기 (초)
DATA_REFRESH_CLOSE = 60 * 30      # 장외 시세 갱신 주기
CHART_CACHE_TTL    = 60 * 60      # 주봉 캐시 유지 시간 (1시간)
PAGE_DURATION      = 20           # 페이지 자동 전환 주기 (초)
FULL_REFRESH_EVERY = 30           # 부분 갱신 N회마다 전체 갱신

FONT_REGULAR = '/usr/share/fonts/truetype/nanum/NanumGothic.ttf'
FONT_BOLD    = '/usr/share/fonts/truetype/nanum/NanumGothicBold.ttf'

MARKET_OPEN  = dtime(9, 0)
MARKET_CLOSE = dtime(15, 30)

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s'
)
log = logging.getLogger(__name__)


# ─────────────────────────────────────────────────────────────
# 데이터 조회
# ─────────────────────────────────────────────────────────────
def fetch_stock(code):
    """현재가 (네이버 모바일 API)"""
    url = f'https://m.stock.naver.com/api/stock/{code}/basic'
    headers = {
        'User-Agent': 'Mozilla/5.0 (Linux; Android 10) AppleWebKit/537.36',
        'Referer': 'https://m.stock.naver.com/',
    }
    try:
        r = requests.get(url, headers=headers, timeout=5)
        r.raise_for_status()
        d = r.json()

        def to_int(s):
            return int(str(s).replace(',', '').replace('+', '').replace('-', '')) if s else 0

        change = to_int(d.get('compareToPreviousClosePrice', '0'))
        sign = d.get('compareToPreviousPrice', {}).get('code', '3')
        if sign in ('4', '5'):
            change = -abs(change)

        return {
            'name'        : d.get('stockName', ''),
            'price'       : to_int(d.get('closePrice', '0')),
            'change'      : change,
            'change_rate' : float(d.get('fluctuationsRatio', '0')),
            'sign'        : sign,
            'open'        : to_int(d.get('openPrice', '0')),
            'high'        : to_int(d.get('highPrice', '0')),
            'low'         : to_int(d.get('lowPrice', '0')),
        }
    except Exception as e:
        log.error(f"[{code}] 시세 조회 실패: {e}")
        return None


def fetch_weekly_chart(code, weeks=CHART_WEEKS):
    """
    주봉 종가 리스트 [(YYYYMMDD, close), ...]
    네이버 차트 API: 응답이 Python list-literal 형태의 텍스트
    """
    end = datetime.now()
    # 여유분 포함해서 weeks*7+30일치 요청
    start = end - timedelta(days=weeks * 7 + 30)
    url = (f'https://api.finance.naver.com/siseJson.naver'
           f'?symbol={code}&requestType=1'
           f'&startTime={start.strftime("%Y%m%d")}'
           f'&endTime={end.strftime("%Y%m%d")}'
           f'&timeframe=week')
    try:
        r = requests.get(url, headers={'User-Agent': 'Mozilla/5.0'}, timeout=8)
        r.raise_for_status()
        text = r.text.strip()

        # 각 [...] 블록 추출 후 파싱
        rows = re.findall(r"\[([^\[\]]+)\]", text)
        data = []
        for row in rows:
            parts = [p.strip() for p in row.split(',')]
            if len(parts) < 5:
                continue
            date_str = parts[0].strip("'\" ")
            if not date_str.isdigit():     # 헤더 행 스킵
                continue
            try:
                close = float(parts[4])
                data.append((date_str, close))
            except (ValueError, IndexError):
                continue

        return data[-weeks:] if len(data) > weeks else data
    except Exception as e:
        log.error(f"[{code}] 주봉 조회 실패: {e}")
        return []


# ─────────────────────────────────────────────────────────────
# 유틸
# ─────────────────────────────────────────────────────────────
def fmt_price(p):
    return f"{p:,}"


def fmt_axis(v):
    """차트 y축 라벨용 포맷"""
    if v >= 10000:
        return f'{v/10000:.1f}만'
    if v >= 1000:
        return f'{v/1000:.1f}k'
    return f'{int(v)}'


def change_symbol(sign):
    if sign in ('1', '2'):
        return '▲'
    if sign in ('4', '5'):
        return '▼'
    return '-'


def is_market_hours():
    now = datetime.now()
    if now.weekday() >= 5:
        return False
    return MARKET_OPEN <= now.time() <= MARKET_CLOSE


# ─────────────────────────────────────────────────────────────
# 화면 그리기
# ─────────────────────────────────────────────────────────────
def _new_canvas(epd):
    """가로 모드 빈 캔버스 (250x122)"""
    width, height = epd.height, epd.width
    image = Image.new('1', (width, height), 255)
    return image, ImageDraw.Draw(image), width, height


def build_summary_image(epd, stocks_data):
    """요약 페이지: 양 종목을 위/아래 분할 표시"""
    image, draw, width, height = _new_canvas(epd)

    f_hdr   = ImageFont.truetype(FONT_BOLD,    11)
    f_time  = ImageFont.truetype(FONT_REGULAR, 10)
    f_name  = ImageFont.truetype(FONT_BOLD,    13)
    f_price = ImageFont.truetype(FONT_BOLD,    22)
    f_chg   = ImageFont.truetype(FONT_BOLD,    11)
    f_tiny  = ImageFont.truetype(FONT_REGULAR,  9)

    market = '장중' if is_market_hours() else '장외'
    draw.text((2, 0), f'주가[{market}]', font=f_hdr, fill=0)
    now_str = datetime.now().strftime('%m-%d %H:%M')
    bbox = draw.textbbox((0, 0), now_str, font=f_time)
    draw.text((width - (bbox[2] - bbox[0]) - 2, 1), now_str, font=f_time, fill=0)
    draw.line([(0, 13), (width, 13)], fill=0, width=1)

    section_h = (height - 14) // len(stocks_data)
    y = 15

    for i, d in enumerate(stocks_data):
        if d is None:
            draw.text((4, y + 18), '데이터 조회 실패', font=f_name, fill=0)
            y += section_h
            continue

        draw.text((3, y + 2), d['name'], font=f_name, fill=0)
        price_str = fmt_price(d['price'])
        bbox = draw.textbbox((0, 0), price_str, font=f_price)
        draw.text((width - (bbox[2]-bbox[0]) - 3, y + 1), price_str, font=f_price, fill=0)

        sym = change_symbol(d['sign'])
        chg_str = f"{sym} {abs(d['change']):,} ({d['change_rate']:+.2f}%)"
        draw.text((3, y + 26), chg_str, font=f_chg, fill=0)

        ohl = f"시{fmt_price(d['open'])} 고{fmt_price(d['high'])} 저{fmt_price(d['low'])}"
        draw.text((3, y + 41), ohl, font=f_tiny, fill=0)

        y += section_h
        if i < len(stocks_data) - 1:
            draw.line([(0, y - 1), (width, y - 1)], fill=0, width=1)

    return image


def build_chart_image(epd, stock, chart_data):
    """차트 페이지: 단일 종목 + 주봉 라인 차트"""
    image, draw, width, height = _new_canvas(epd)

    f_name  = ImageFont.truetype(FONT_BOLD,    13)
    f_price = ImageFont.truetype(FONT_BOLD,    16)
    f_chg   = ImageFont.truetype(FONT_BOLD,    11)
    f_tiny  = ImageFont.truetype(FONT_REGULAR,  9)

    if stock is None:
        draw.text((60, 50), '데이터 조회 실패', font=f_name, fill=0)
        return image

    # 헤더: 종목명 + 현재가
    draw.text((3, 0), stock['name'], font=f_name, fill=0)
    price_str = fmt_price(stock['price'])
    bbox = draw.textbbox((0, 0), price_str, font=f_price)
    draw.text((width - (bbox[2]-bbox[0]) - 3, 0), price_str, font=f_price, fill=0)

    # 등락 + 페이지 라벨
    sym = change_symbol(stock['sign'])
    chg_str = f"{sym} {abs(stock['change']):,} ({stock['change_rate']:+.2f}%)"
    draw.text((3, 18), chg_str, font=f_chg, fill=0)

    label = f'주봉 {len(chart_data)}주'
    bbox = draw.textbbox((0, 0), label, font=f_tiny)
    draw.text((width - (bbox[2]-bbox[0]) - 3, 21), label, font=f_tiny, fill=0)

    draw.line([(0, 32), (width, 32)], fill=0, width=1)

    # ── 차트 영역 ──
    cx, cy = 3, 36                  # 좌상단
    cw, ch = 200, 82                 # 폭, 높이 (오른쪽에 라벨 공간 확보)

    if not chart_data or len(chart_data) < 2:
        draw.text((cx + 40, cy + ch//2 - 5), '차트 데이터 없음', font=f_tiny, fill=0)
        return image

    closes = [c[1] for c in chart_data]
    pmin, pmax = min(closes), max(closes)
    if pmax == pmin:
        pmax = pmin + 1
    span = pmax - pmin

    # 축
    draw.line([(cx, cy), (cx, cy + ch)], fill=0, width=1)
    draw.line([(cx, cy + ch), (cx + cw, cy + ch)], fill=0, width=1)

    # 가로 그리드 (중간선)
    mid_y = cy + ch // 2
    for x in range(cx + 2, cx + cw, 6):
        draw.point((x, mid_y), fill=0)

    # 라인 그리기
    n = len(closes)
    pts = []
    for i, c in enumerate(closes):
        px = cx + 1 + int(i * (cw - 2) / max(n - 1, 1))
        py = cy + ch - 1 - int((c - pmin) / span * (ch - 3))
        pts.append((px, py))

    for i in range(len(pts) - 1):
        draw.line([pts[i], pts[i + 1]], fill=0, width=1)

    # 현재가(마지막 봉) 강조 표시
    last_x, last_y = pts[-1]
    draw.ellipse([last_x - 2, last_y - 2, last_x + 2, last_y + 2], fill=0)

    # Y축 라벨 (오른쪽)
    label_x = cx + cw + 3
    draw.text((label_x, cy - 4),         fmt_axis(pmax), font=f_tiny, fill=0)
    draw.text((label_x, mid_y - 5),      fmt_axis((pmax + pmin) / 2), font=f_tiny, fill=0)
    draw.text((label_x, cy + ch - 9),    fmt_axis(pmin), font=f_tiny, fill=0)

    # X축 라벨 (시작 / 끝 날짜)
    if chart_data:
        d_start = chart_data[0][0]
        d_end   = chart_data[-1][0]
        # YYYYMMDD → MM/DD
        s_str = f'{d_start[4:6]}/{d_start[6:8]}'
        e_str = f'{d_end[4:6]}/{d_end[6:8]}'
        draw.text((cx + 2, cy + ch + 1), s_str, font=f_tiny, fill=0)
        bbox = draw.textbbox((0, 0), e_str, font=f_tiny)
        draw.text((cx + cw - (bbox[2]-bbox[0]) - 2, cy + ch + 1), e_str, font=f_tiny, fill=0)

    return image


# ─────────────────────────────────────────────────────────────
# 메인 루프
# ─────────────────────────────────────────────────────────────
running = True

def handle_signal(signum, frame):
    global running
    log.info(f"신호 수신({signum}) - 종료 처리")
    running = False


def render(epd, image, state):
    """전체/부분 갱신을 자동 선택해 화면 출력"""
    if state['first'] or state['partial_count'] >= FULL_REFRESH_EVERY:
        epd.init()
        epd.display(epd.getbuffer(image))
        state['partial_count'] = 0
        state['first'] = False
        log.info("전체 갱신")
    else:
        try:
            epd.displayPartial(epd.getbuffer(image))
            state['partial_count'] += 1
            log.info(f"부분 갱신 ({state['partial_count']}/{FULL_REFRESH_EVERY})")
        except AttributeError:
            epd.display(epd.getbuffer(image))
            log.info("전체 갱신 (부분 갱신 미지원)")


def main():
    signal.signal(signal.SIGINT,  handle_signal)
    signal.signal(signal.SIGTERM, handle_signal)

    log.info("주가 모니터 시작 (Waveshare 2.13\")")
    epd = epd_module.EPD()
    epd.init()
    epd.Clear(0xFF)

    stocks_data = [None] * len(STOCKS)
    chart_cache = {}                 # code -> ([(date, close), ...], fetched_at)
    state = {'first': True, 'partial_count': 0}

    total_pages = 1 + len(STOCKS)    # 요약 + 종목별 차트
    current_page = 0

    last_data_fetch = 0
    last_page_switch = time.time()

    try:
        while running:
            now = time.time()
            need_redraw = False

            # ── 시세 갱신 ──
            interval = DATA_REFRESH_OPEN if is_market_hours() else DATA_REFRESH_CLOSE
            if now - last_data_fetch >= interval:
                for i, s in enumerate(STOCKS):
                    d = fetch_stock(s['code'])
                    stocks_data[i] = d
                    if d:
                        log.info(f"{d['name']:8s} {d['price']:>8,}  "
                                 f"{change_symbol(d['sign'])}{abs(d['change']):>6,}  "
                                 f"({d['change_rate']:+.2f}%)")
                last_data_fetch = now
                need_redraw = True

            # ── 주봉 캐시 갱신 ──
            for s in STOCKS:
                code = s['code']
                cached = chart_cache.get(code)
                if cached is None or now - cached[1] >= CHART_CACHE_TTL:
                    log.info(f"[{code}] 주봉 조회")
                    chart = fetch_weekly_chart(code)
                    chart_cache[code] = (chart, now)
                    need_redraw = True

            # ── 페이지 전환 ──
            if now - last_page_switch >= PAGE_DURATION:
                current_page = (current_page + 1) % total_pages
                last_page_switch = now
                need_redraw = True
                log.info(f"페이지 전환 → {current_page}")

            # ── 그리기 ──
            if need_redraw:
                if current_page == 0:
                    image = build_summary_image(epd, stocks_data)
                else:
                    idx = current_page - 1
                    code = STOCKS[idx]['code']
                    chart = chart_cache.get(code, ([], 0))[0]
                    image = build_chart_image(epd, stocks_data[idx], chart)
                render(epd, image, state)

            # 1초 단위로 깨어나서 페이지/데이터 시점 체크
            for _ in range(1):
                if not running:
                    break
                time.sleep(1)

    except Exception as e:
        log.exception(f"오류: {e}")
    finally:
        log.info("e-Paper 정리")
        try:
            epd.init()
            epd.Clear(0xFF)
            epd.sleep()
        except Exception:
            pass
        log.info("종료 완료")


if __name__ == '__main__':
    main()

4. 실행 테스트

python3 stock_monitor.py

5. 부팅 시 자동 실행 (systemd) /etc/systemd/system/stock-monitor.service 파일 생성:

[Unit]
Description=Stock Price e-Paper Monitor
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=pi
WorkingDirectory=/home/pi/stock_monitor
ExecStart=/usr/bin/python3 /home/pi/stock_monitor/stock_monitor.py
Restart=on-failure
RestartSec=10

[Install]
WantedBy=multi-user.target

활성화

sudo systemctl daemon-reload
sudo systemctl enable stock-monitor.service
sudo systemctl start stock-monitor.service
sudo systemctl status stock-monitor.service

로그확인

journalctl -u stock-monitor.service -f