다른 명령
| 85번째 줄: | 85번째 줄: | ||
<source lang=python> | <source lang=python> | ||
journalctl -u stock-monitor.service -f | journalctl -u stock-monitor.service -f | ||
</source> | |||
=== 삼성전자 , 현대차 주가 표시 === | |||
<source lang=python> | |||
#!/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() | |||
</source> | </source> | ||
2026년 5월 6일 (수) 23:17 판
개요
데이터는 네이버 금융 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. 실행 테스트
python3 stock_monitor.py
4. 부팅 시 자동 실행 (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
삼성전자 , 현대차 주가 표시
#!/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()