|
|
| 46번째 줄: |
46번째 줄: |
| </source> | | </source> |
|
| |
|
| 3. 프로그램 코드 | | 3. 실행 테스트 |
| | <source lang=python> |
| | python3 stock_monitor.py |
| | </source> |
|
| |
|
| === 삼성전자 , 현대차 주가 표시 ===
| | 4. 부팅 시 자동 실행 (systemd) |
| | /etc/systemd/system/stock-monitor.service 파일 생성: |
| <source lang=python> | | <source lang=python> |
| #!/usr/bin/env python3
| | [Unit] |
| # -*- coding: utf-8 -*-
| | Description=Stock Price e-Paper Monitor |
| """
| | After=network-online.target |
| 삼성전자(005930) / 현대차(005380) 실시간 주가 모니터 + 주봉 차트
| | Wants=network-online.target |
| Raspberry Pi + Waveshare 2.13" e-Paper (250x122)
| |
|
| |
|
| 페이지 자동 전환:
| | [Service] |
| Page 0 - 요약 (양 종목)
| | Type=simple |
| Page 1 - 삼성전자 + 주봉 차트
| | User=pi |
| Page 2 - 현대차 + 주봉 차트
| | WorkingDirectory=/home/pi/stock_monitor |
| | ExecStart=/usr/bin/python3 /home/pi/stock_monitor/stock_monitor.py |
| | Restart=on-failure |
| | RestartSec=10 |
|
| |
|
| Author: 치치 (dbaworks)
| | [Install] |
| """
| | WantedBy=multi-user.target |
|
| |
|
| import os
| | </source> |
| 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):
| | <source lang=bash> |
| sys.path.append(libdir)
| | sudo systemctl daemon-reload |
| | sudo systemctl enable stock-monitor.service |
| | sudo systemctl start stock-monitor.service |
| | sudo systemctl status stock-monitor.service |
| | </source> |
|
| |
|
| 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 lang=python> |
| | journalctl -u stock-monitor.service -f |
| </source> | | </source> |
