한 줄 요약
오렌지파이5에서 **보유 종목(주식/ETF)**을 파일로 관리하고,
- 수익 최고점(peak) 대비 -X% 트레일링
- 매수가 대비 -Y% 손절
- 손절/트레일링 근접 경고
- 1시간 요약 / 15:35 요약 / 16:00 리포트
- 리스크 HIGH일 때만 30분→10분 자동 단축
- 다음날 자동 재감시(active=1 복구)
를 시놀로지 Chat 웹훅으로 자동 알림해주는 파이썬 봇입니다.
※ 가격은 yfinance 기반이라 “지연 Close”로 동작합니다(실시간 아님).
실시간으로 바꾸려면 추후 키움 API 등 연동이 필요합니다.
1) 동작 개념(핵심 로직)
✅ 트레일링(수익 구간)
- 내 매수금액(평가금액 기준) 대비 최고점(peak)을 계속 갱신
- 기준선 =
peak × (1 - trailing%) - 현재가가 기준선 이하로 내려오면 매도 알림
✅ 가변 트레일링(수익이 커질수록 더 타이트)
예: base trailing이 10%인 종목이라면
- 수익 0~20%: 10%
- 수익 20~50%: 7% (10%×0.7)
- 수익 50% 이상: 5% (10%×0.5)
✅ 손절(매수가 기준)
- 손절선 =
매수금액 × (1 - stop_loss%) - 현재가가 손절선 이하로 내려오면 손절 알림
✅ 근접 경고
- 트레일링 기준선까지 2% 이내로 접근하면 “근접 경고”
- 손절선까지 2% 이내로 접근하면 “손절 근접 경고”
✅ 리스크 레벨(LOW / MEDIUM / HIGH)
- Trigger(기준선 이하) / Near(근접) 종목 수
- 위험 노출 비중(Trigger+Near 종목 평가금액 합 / 전체 평가금액)
을 기준으로 LOW/MEDIUM/HIGH 표시
→ 장중 요약, 16:00 리포트에 같이 출력
✅ 실행 주기 자동 변경(장중만)
- 평소 장중: 30분
- 리스크 HIGH: 10분(더 촘촘히 감시)
- 장외: 120분(불필요 호출 절약)
✅ 다음날 자동 재활성화
- 매도/손절 알림 후 active=0으로 자동 비활성화(옵션)
- 다음날 09:01에 자동으로 active=1로 복구(옵션)
2) 폴더 구성(그대로 만들기)
아래 3개 파일만 있으면 됩니다.
jusik_top10/
├─ main.py
├─ config.json
├─ positions.csv
└─ state.json (없으면 자동 생성됨, 처음엔 {} 권장)
3) 설치(오렌지파이5)
sudo apt update
sudo apt install -y python3-pip
python3 -m pip install --user requests yfinance
4) config.json (복붙용)
synology_chat_webhook_url만 본인 값으로 바꿔주세요.
{
"synology_chat_webhook_url": "여기에_시놀로지챗_웹훅_URL_붙여넣기", "default_trailing_pct": 0.10,
"default_stop_loss_pct": 0.08, "market_open": "09:00",
"market_close": "15:30", "startup_notify_enabled": true, "near_alert_enabled": true,
"near_alert_distance_pct": 2.0,
"near_alert_cooldown_minutes": 60, "stoploss_near_enabled": true,
"stoploss_near_distance_pct": 2.0,
"stoploss_near_cooldown_minutes": 180, "auto_disable_on_alert": true, "hourly_summary_enabled": true,
"hourly_summary_interval_minutes": 60, "daily_summary_time": "15:35", "after_close_summary_enabled": true,
"after_close_summary_time": "16:00",
"after_close_include_inactive": false, "adaptive_trailing_enabled": true,
"adaptive_trailing_factors": [
{ "min_profit_pct": 0, "factor": 1.0 },
{ "min_profit_pct": 20, "factor": 0.7 },
{ "min_profit_pct": 50, "factor": 0.5 }
], "risk_high_near_count": 3,
"risk_high_near_exposure_pct": 30, "adaptive_interval_enabled": true,
"interval_minutes_market": 30,
"interval_minutes_high_risk": 10,
"interval_minutes_off_market": 120, "daily_reactivate_enabled": true,
"daily_reactivate_time": "09:01"
}
5) positions.csv (복붙용)
market: KS(코스피), KQ(코스닥)trailing_pct: 0.10 = 10%stop_loss_pct: 0.08 = -8%active: 1(감시), 0(감시중지)
code,name,market,qty,buy_price,trailing_pct,stop_loss_pct,active
005930,삼성전자,KS,10,70000,0.10,0.08,1
069500,KODEX 200,KS,20,35000,0.08,0.07,1
6) state.json 초기화(권장)
처음 적용할 때 state가 꼬였으면 초기화가 안전합니다.
echo "{}" > state.json
7) main.py (전체 소스 — 그대로 복붙)
아래 코드를
main.py로 저장하세요.
# (아래 전체 코드 그대로 복붙)
"""
Orange Pi 5 (Ubuntu/ARM) 주식/ETF 트레일링 스탑 알림 시스템 (yfinance 기반, 지연 Close OK)[핵심 기능]
- positions.csv 관리: 종목/수량/매수가/trailing_pct/stop_loss_pct/active
- state.json 유지: peak_value, last_alerted_peak, 근접 경고 시각, 요약 발송 기록 등
- Synology Chat Incoming Webhook으로 알림 전송[장중(09:00~15:30) 기능]
1) 손절(매수가 기준) 알림: entry*(1-stop_loss_pct) 이하
2) 손절 근접 경고: 손절선까지 +X% 남으면(0% 이하면 손절)
3) 트레일링(수익구간) 알림: peak*(1-eff_trailing) 이하
4) 트레일링 근접 경고: 기준선까지 +X% 남으면
5) 1시간 요약(장중): 총 수익률 + 종목별 요약 + 리스크 레벨[장마감 이후]
- 15:35 일일 요약(TOP3/WORST3)
- 16:00 리포트(섹션 분리: TRIGGER/NEAR/SAFE + 손절 정보 + 리스크)[추가 보강]
- 가변 트레일링(Adaptive Trailing): base_trailing_pct × factor(수익구간별)
- 리스크 HIGH일 때 장중 주기 자동 단축(예: 30→10분)
- 매도/손절 알림 후 auto_disable_on_alert=true면 active=0 자동 변경
- 다음날 지정 시간에 active=0 → active=1 자동 재활성화(옵션)
- state.json 깨짐 자동복구 + 원자저장
- positions.csv 원자저장
- 시작 1회 안내 메시지[주의]
- yfinance는 지연 Close 기반(실시간 아님)
"""import os
import json
import time
import csv
import tempfile
from dataclasses import dataclass
from datetime import datetime, time as dtimeimport requests
import yfinance as yfBASE = os.path.dirname(os.path.abspath(__file__))
CONFIG_PATH = os.path.join(BASE, "config.json")
POSITIONS_PATH = os.path.join(BASE, "positions.csv")
STATE_PATH = os.path.join(BASE, "state.json")def log(step: str, msg: str = "", level: str = "INFO"):
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
if msg:
print(f"[{now}] [{level}] [{step}] {msg}")
else:
print(f"[{now}] [{level}] [{step}]")def fmt_money(x) -> str:
return f"{x:,.0f}"def parse_hhmm(s: str) -> dtime:
hh, mm = s.split(":")
return dtime(hour=int(hh), minute=int(mm))def is_market_open(now: datetime, open_t: dtime, close_t: dtime) -> bool:
t = now.time()
return open_t <= t <= close_tdef load_json_safe(path, default):
if not os.path.exists(path):
return default
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except json.JSONDecodeError as e:
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
bak = f"{path}.broken_{ts}"
try:
os.replace(path, bak)
except:
try:
import shutil
shutil.copy2(path, bak)
except:
pass
print(f"[WARN] JSON이 깨져 자동 복구했습니다. 백업: {bak} / 오류: {e}")
return defaultdef save_json_atomic(path, data):
tmp_dir = os.path.dirname(path) or "."
fd, tmp_path = tempfile.mkstemp(prefix="state_", suffix=".json", dir=tmp_dir)
os.close(fd)
try:
with open(tmp_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
os.replace(tmp_path, path)
finally:
if os.path.exists(tmp_path):
try:
os.remove(tmp_path)
except:
passdef parse_float_optional(x):
if x is None:
return None
s = str(x).strip()
if s == "":
return None
return float(s)def parse_int_optional(x, default=1):
if x is None:
return default
s = str(x).strip()
if s == "":
return default
try:
return int(float(s))
except:
return default@dataclass
class Position:
code: str
name: str
market: str
qty: float
buy_price: float
trailing_pct: float
stop_loss_pct: float
active: int @property
def ticker(self) -> str:
return f"{self.code}.{self.market.upper()}"def load_positions_csv(path, default_trailing_pct: float, default_stop_loss_pct: float) -> list[Position]:
if not os.path.exists(path):
raise RuntimeError(f"positions.csv가 없습니다: {path}") positions = []
with open(path, "r", encoding="utf-8-sig") as f:
reader = csv.DictReader(f)
required = {"code", "qty", "buy_price"}
if not required.issubset(set(reader.fieldnames or [])):
raise RuntimeError(f"positions.csv 컬럼이 필요합니다: {sorted(required)}") for row in reader:
code = row["code"].strip()
name = (row.get("name") or code).strip()
market = (row.get("market") or "KS").strip().upper()
if market not in ("KS", "KQ"):
raise RuntimeError(f"{code} market 값 오류: {market} (KS/KQ만 허용)") qty = float(row["qty"])
buy_price = float(row["buy_price"]) trailing_pct = parse_float_optional(row.get("trailing_pct"))
if trailing_pct is None:
trailing_pct = default_trailing_pct
if not (0 < trailing_pct < 1):
raise RuntimeError(f"{code} trailing_pct 비정상: {trailing_pct}") stop_loss_pct = parse_float_optional(row.get("stop_loss_pct"))
if stop_loss_pct is None:
stop_loss_pct = default_stop_loss_pct
if not (0 <= stop_loss_pct < 1):
raise RuntimeError(f"{code} stop_loss_pct 비정상: {stop_loss_pct}") active = parse_int_optional(row.get("active"), default=1)
active = 1 if active != 0 else 0 positions.append(Position(code, name, market, qty, buy_price, trailing_pct, stop_loss_pct, active))
return positionsdef save_positions_csv_atomic(path, positions: list[Position]):
fieldnames = ["code", "name", "market", "qty", "buy_price", "trailing_pct", "stop_loss_pct", "active"]
tmp_dir = os.path.dirname(path) or "."
fd, tmp_path = tempfile.mkstemp(prefix="positions_", suffix=".csv", dir=tmp_dir)
os.close(fd)
try:
with open(tmp_path, "w", encoding="utf-8-sig", newline="") as f:
w = csv.DictWriter(f, fieldnames=fieldnames)
w.writeheader()
for p in positions:
w.writerow({
"code": p.code, "name": p.name, "market": p.market,
"qty": p.qty, "buy_price": p.buy_price,
"trailing_pct": p.trailing_pct, "stop_loss_pct": p.stop_loss_pct,
"active": p.active
})
os.replace(tmp_path, path)
finally:
if os.path.exists(tmp_path):
try: os.remove(tmp_path)
except: passdef send_synology_chat(webhook_url: str, text: str):
payload_json = json.dumps({"text": text}, ensure_ascii=False)
r = requests.post(webhook_url, data={"payload": payload_json}, timeout=10)
r.raise_for_status()def get_last_price_yf(ticker: str) -> float:
hist = yf.Ticker(ticker).history(period="7d")
if hist is None or hist.empty:
raise RuntimeError(f"가격 데이터 비어있음: {ticker}")
return float(hist["Close"].iloc[-1])def get_adaptive_factor(cfg: dict, peak_profit_pct: float) -> float:
if not bool(cfg.get("adaptive_trailing_enabled", True)):
return 1.0
tiers = cfg.get("adaptive_trailing_factors", [])
if not isinstance(tiers, list) or len(tiers) == 0:
return 1.0 valid = []
for t in tiers:
try:
m = float(t.get("min_profit_pct", 0))
f = float(t.get("factor", 1.0))
if f > 0:
valid.append((m, f))
except:
pass if not valid:
return 1.0 valid.sort(key=lambda x: x[0])
chosen = valid[0][1]
for m, f in valid:
if peak_profit_pct >= m:
chosen = f
else:
break
return chosendef effective_trailing_pct(cfg: dict, base_trailing_pct: float, entry_value: float, peak_value: float) -> float:
if entry_value <= 0:
return base_trailing_pct
peak_profit_pct = (peak_value - entry_value) / entry_value * 100.0
factor = get_adaptive_factor(cfg, peak_profit_pct)
eff = base_trailing_pct * factor
eff = max(eff, 0.01)
eff = min(eff, base_trailing_pct)
return effdef cleanup_state(state: dict, active_codes: set[str]):
for k in list(state.keys()):
if k in ("_summary", "_hourly_summary", "_startup", "_after_close_summary", "_reactivate"):
continue
if k not in active_codes:
del state[k]def should_send_daily_summary(state: dict, now: datetime, summary_time: dtime) -> bool:
if now.time() < summary_time:
return False
today = now.strftime("%Y-%m-%d")
s = state.setdefault("_summary", {})
return s.get("last_date") != todaydef should_send_hourly_summary_market_only(state: dict, now: datetime, interval_minutes: int, market_open: bool) -> bool:
if not market_open:
return False
hourly = state.setdefault("_hourly_summary", {})
last_str = hourly.get("last_sent")
if not last_str:
return True
last_dt = datetime.strptime(last_str, "%Y-%m-%d %H:%M:%S")
diff_min = (now - last_dt).total_seconds() / 60.0
return diff_min >= interval_minutesdef should_send_after_close_summary(state: dict, now: datetime, summary_time: dtime) -> bool:
if now.time() < summary_time:
return False
today = now.strftime("%Y-%m-%d")
s = state.setdefault("_after_close_summary", {})
return s.get("last_date") != todaydef should_send_startup_notify(state: dict) -> bool:
st = state.setdefault("_startup", {})
return not bool(st.get("sent", False))def mark_startup_notify_sent(state: dict):
st = state.setdefault("_startup", {})
st["sent"] = True
st["sent_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")def should_reactivate_today(state: dict, now: datetime, hhmm: str) -> bool:
t = parse_hhmm(hhmm)
if now.time() < t:
return False
today = now.strftime("%Y-%m-%d")
s = state.setdefault("_reactivate", {})
return s.get("last_date") != todaydef reactivate_all_positions(positions_all: list[Position], state: dict) -> bool:
changed = False
for p in positions_all:
if p.active == 0:
p.active = 1
changed = True
if changed:
state.setdefault("_reactivate", {})["last_date"] = datetime.now().strftime("%Y-%m-%d")
return changeddef calc_risk_level(detail_list: list[dict], total_current_active: float, cfg: dict, near_dist: float) -> dict:
trigger = [d for d in detail_list if d.get("active") == 1 and d.get("to_stop", 999) <= 0]
near = [d for d in detail_list if d.get("active") == 1 and 0 < d.get("to_stop", 999) <= near_dist] trigger_cnt = len(trigger)
near_cnt = len(near) risk_exposure = sum(float(d.get("cur", 0.0)) for d in (trigger + near))
exposure_pct = (risk_exposure / total_current_active * 100.0) if total_current_active > 0 else 0.0 high_near_cnt = int(cfg.get("risk_high_near_count", 3))
high_exposure_pct = float(cfg.get("risk_high_near_exposure_pct", 30)) if trigger_cnt > 0 or near_cnt >= high_near_cnt or exposure_pct >= high_exposure_pct:
level = "HIGH"
elif near_cnt > 0:
level = "MEDIUM"
else:
level = "LOW" return {
"level": level,
"trigger_cnt": trigger_cnt,
"near_cnt": near_cnt,
"risk_exposure": risk_exposure,
"exposure_pct": exposure_pct
}def top_worst_lines(rows, n=3):
rows_desc = sorted(rows, key=lambda r: r["rate_now"], reverse=True)[:n]
rows_asc = sorted(rows, key=lambda r: r["rate_now"], reverse=False)[:n] def line(r):
sign = "+" if r["profit_now"] >= 0 else ""
return f"- {r['name']}({r['code']}): {r['rate_now']:+.2f}% ({sign}{fmt_money(r['profit_now'])}원)" return [line(r) for r in rows_desc], [line(r) for r in rows_asc]def run_once(cfg: dict, positions_all: list[Position], state: dict):
now = datetime.now()
now_str = now.strftime("%Y-%m-%d %H:%M:%S") webhook_url = cfg["synology_chat_webhook_url"] market_open_t = parse_hhmm(cfg.get("market_open", "09:00"))
market_close_t = parse_hhmm(cfg.get("market_close", "15:30"))
daily_summary_t = parse_hhmm(cfg.get("daily_summary_time", "15:35")) hourly_enabled = bool(cfg.get("hourly_summary_enabled", True))
hourly_interval = int(cfg.get("hourly_summary_interval_minutes", 60)) auto_disable = bool(cfg.get("auto_disable_on_alert", True)) near_enabled = bool(cfg.get("near_alert_enabled", True))
near_dist = float(cfg.get("near_alert_distance_pct", 2.0))
near_cooldown = int(cfg.get("near_alert_cooldown_minutes", 60)) sl_near_enabled = bool(cfg.get("stoploss_near_enabled", True))
sl_near_dist = float(cfg.get("stoploss_near_distance_pct", 2.0))
sl_near_cooldown = int(cfg.get("stoploss_near_cooldown_minutes", 180)) after_close_enabled = bool(cfg.get("after_close_summary_enabled", True))
after_close_time = parse_hhmm(cfg.get("after_close_summary_time", "16:00"))
include_inactive = bool(cfg.get("after_close_include_inactive", False)) positions_active = [p for p in positions_all if p.active == 1]
active_codes = {p.code for p in positions_active}
cleanup_state(state, active_codes) market_open = is_market_open(now, market_open_t, market_close_t)
log("MARKET", f"장중={market_open}") price_targets = positions_all if include_inactive else positions_active
price_cache: dict[str, float] = {} log("PRICE", f"조회 대상={len(price_targets)}")
for p in price_targets:
if p.code in price_cache:
continue
try:
price_cache[p.code] = get_last_price_yf(p.ticker)
except Exception as e:
log("PRICE", f"FAIL {p.name}({p.code}) {e}", level="WARN") positions_ok = [p for p in positions_active if p.code in price_cache]
if not positions_ok:
return False, "LOW", market_open total_entry = total_current = total_peak = 0.0
rows = [] for p in positions_ok:
entry_value = p.qty * p.buy_price
current_value = p.qty * price_cache[p.code]
s = state.setdefault(p.code, {}) peak_value = float(s.get("peak_value", entry_value))
if current_value > peak_value:
peak_value = current_value
s["peak_value"] = peak_value profit_now = current_value - entry_value
rate_now = (profit_now / entry_value * 100.0) if entry_value else 0.0 total_entry += entry_value
total_current += current_value
total_peak += peak_value rows.append({"code": p.code, "name": p.name, "profit_now": profit_now, "rate_now": rate_now}) total_profit_now = total_current - total_entry
total_rate_now = (total_profit_now / total_entry * 100.0) if total_entry else 0.0 changed_positions = False
if market_open:
pos_map = {p.code: p for p in positions_all} for p in positions_ok:
entry_value = p.qty * p.buy_price
current_value = p.qty * price_cache[p.code]
s = state.setdefault(p.code, {})
peak_value = float(s.get("peak_value", entry_value)) # 손절 근접 + 손절 트리거
if p.stop_loss_pct > 0:
stop_loss_value = entry_value * (1 - p.stop_loss_pct)
to_stoploss = (current_value - stop_loss_value) / stop_loss_value * 100.0 if stop_loss_value > 0 else None if sl_near_enabled and to_stoploss is not None and (0 < to_stoploss <= sl_near_dist):
ns = state.setdefault(p.code, {})
last_sl_near = ns.get("last_stoploss_near_alert_at")
can_send = True
if last_sl_near:
last_dt = datetime.strptime(last_sl_near, "%Y-%m-%d %H:%M:%S")
can_send = (now - last_dt).total_seconds() >= sl_near_cooldown * 60
if can_send:
msg = (
f"[손절 근접 경고/매수가 -{int(p.stop_loss_pct*100)}%]\n"
f"- {p.name}({p.code})\n"
f"- 손절선까지 거리: {to_stoploss:+.2f}% (0% 이하이면 손절)\n"
f"- 최근가(지연 Close): {fmt_money(price_cache[p.code])}원\n"
f"- 손절선: {fmt_money(stop_loss_value)}원\n"
f"- 평가금액: {fmt_money(current_value)}원\n"
f"- 시각: {now_str}"
)
send_synology_chat(webhook_url, msg)
ns["last_stoploss_near_alert_at"] = now_str stop_loss_triggered = current_value <= stop_loss_value
last_sl = s.get("last_stop_loss_alert_at")
already_today = bool(last_sl and last_sl[:10] == now_str[:10])
if stop_loss_triggered and not already_today:
msg = (
f"[손절 알림/매수가 -{int(p.stop_loss_pct*100)}%]\n"
f"- {p.name}({p.code})\n"
f"- 최근가(지연 Close): {fmt_money(price_cache[p.code])}원\n"
f"- 평가금액: {fmt_money(current_value)}원\n"
f"- 손절선: {fmt_money(stop_loss_value)}원\n"
f"- 매수금액: {fmt_money(entry_value)}원\n"
f"- 시각: {now_str}"
)
send_synology_chat(webhook_url, msg)
s["last_stop_loss_alert_at"] = now_str
if auto_disable and pos_map[p.code].active != 0:
pos_map[p.code].active = 0
changed_positions = True # 트레일링 근접 + 트리거
last_alerted_peak = float(s.get("last_alerted_peak", 0))
if peak_value > entry_value:
eff_trailing = effective_trailing_pct(cfg, p.trailing_pct, entry_value, peak_value)
stop_value = peak_value * (1 - eff_trailing)
to_stop = (current_value - stop_value) / stop_value * 100.0
drop_from_peak = (peak_value - current_value) / peak_value * 100.0
triggered = current_value <= stop_value if near_enabled and (0 < to_stop <= near_dist):
ns = state.setdefault(p.code, {})
last_near = ns.get("last_near_alert_at")
can_send = True
if last_near:
last_dt = datetime.strptime(last_near, "%Y-%m-%d %H:%M:%S")
can_send = (now - last_dt).total_seconds() >= near_cooldown * 60
if can_send:
msg = (
f"[근접 경고/트레일링 {int(eff_trailing*100)}%]\n"
f"- {p.name}({p.code})\n"
f"- 기준선까지 거리: {to_stop:+.2f}% (0% 이하이면 트리거)\n"
f"- 피크 대비 하락률: {drop_from_peak:.2f}%\n"
f"- 최근가(지연 Close): {fmt_money(price_cache[p.code])}원\n"
f"- 시각: {now_str}"
)
send_synology_chat(webhook_url, msg)
ns["last_near_alert_at"] = now_str if triggered and last_alerted_peak < peak_value:
msg = (
f"[매도 알림/트레일링 {int(eff_trailing*100)}%]\n"
f"- {p.name}({p.code})\n"
f"- 기준선까지 거리: {to_stop:+.2f}% (트리거)\n"
f"- 피크 대비 하락률: {drop_from_peak:.2f}%\n"
f"- 최근가(지연 Close): {fmt_money(price_cache[p.code])}원\n"
f"- 시각: {now_str}"
)
send_synology_chat(webhook_url, msg)
s["last_alerted_peak"] = peak_value
if auto_disable and pos_map[p.code].active != 0:
pos_map[p.code].active = 0
changed_positions = True # 리스크/리포트용 상세 리스트
positions_for_report = positions_all if include_inactive else positions_active
detail_list = []
for p in positions_for_report:
if p.code not in price_cache:
continue
entry_value = p.qty * p.buy_price
current_value = p.qty * price_cache[p.code]
s = state.get(p.code, {})
peak_value = float(s.get("peak_value", entry_value))
eff_trailing = effective_trailing_pct(cfg, p.trailing_pct, entry_value, peak_value) if peak_value > entry_value else p.trailing_pct
stop_value = peak_value * (1 - eff_trailing) if peak_value else 0.0
to_stop = (current_value - stop_value) / stop_value * 100.0 if stop_value else 0.0 stop_loss_value = entry_value * (1 - p.stop_loss_pct) if p.stop_loss_pct > 0 else 0.0
to_stoploss = (current_value - stop_loss_value) / stop_loss_value * 100.0 if stop_loss_value > 0 else None profit_now = current_value - entry_value
rate_now = (profit_now / entry_value * 100.0) if entry_value else 0.0 detail_list.append({
"name": p.name, "code": p.code, "active": p.active,
"cur": current_value, "entry": entry_value,
"profit": profit_now, "rate": rate_now,
"peak": peak_value, "eff_trailing": eff_trailing,
"stop": stop_value, "to_stop": to_stop,
"stop_loss_pct": p.stop_loss_pct, "stop_loss_value": stop_loss_value, "to_stoploss": to_stoploss
}) risk = calc_risk_level(detail_list, total_current, cfg, near_dist) # 장중 1시간 요약
if bool(cfg.get("hourly_summary_enabled", True)):
if should_send_hourly_summary_market_only(state, now, int(cfg.get("hourly_summary_interval_minutes", 60)), market_open):
lines = []
for r in rows:
sign = "+" if r["profit_now"] >= 0 else ""
lines.append(f"- {r['name']}({r['code']}): {r['rate_now']:+.2f}% ({sign}{fmt_money(r['profit_now'])}원)") risk_box = (
f"[리스크 레벨: {risk['level']}]\n"
f"- Trigger: {risk['trigger_cnt']} / Near: {risk['near_cnt']}\n"
f"- 위험노출: {fmt_money(risk['risk_exposure'])}원 ({risk['exposure_pct']:.1f}%)\n"
) msg = (
f"[1시간 포트폴리오 요약]\n"
f"- 총 손익: {fmt_money(total_profit_now)}원\n"
f"- 총 수익률: {total_rate_now:+.2f}%\n\n"
+ risk_box + "\n"
+ "\n".join(lines) +
f"\n\n- 시각: {now_str}"
)
send_synology_chat(webhook_url, msg)
state.setdefault("_hourly_summary", {})["last_sent"] = now_str # 15:35 일일 요약
if should_send_daily_summary(state, now, daily_summary_t):
total_profit_peak = total_peak - total_entry
total_rate_peak = (total_profit_peak / total_entry * 100.0) if total_entry else 0.0
top3, worst3 = top_worst_lines(rows, n=3) msg = (
f"[포트폴리오 일일 요약]\n"
f"- 총 매수금액: {fmt_money(total_entry)}원\n"
f"- 총 평가금액(현재): {fmt_money(total_current)}원\n"
f"- 총 손익(현재): {fmt_money(total_profit_now)}원 ({total_rate_now:+.2f}%)\n"
f"- 총 최고평가(피크): {fmt_money(total_peak)}원\n"
f"- 총 손익(피크): {fmt_money(total_profit_peak)}원 ({total_rate_peak:+.2f}%)\n"
f"\n[손익률 TOP3]\n" + "\n".join(top3) +
f"\n\n[손익률 WORST3]\n" + "\n".join(worst3) +
f"\n\n- 시각: {now_str}"
)
send_synology_chat(webhook_url, msg)
state.setdefault("_summary", {})["last_date"] = now.strftime("%Y-%m-%d") # 16:00 리포트(섹션 분리)
if bool(cfg.get("after_close_summary_enabled", True)) and should_send_after_close_summary(state, now, after_close_time):
trigger = [d for d in detail_list if d["active"] == 1 and d["to_stop"] <= 0]
near = [d for d in detail_list if d["active"] == 1 and 0 < d["to_stop"] <= near_dist]
safe = [d for d in detail_list if d["active"] == 1 and d["to_stop"] > near_dist] def line(d):
a = "A1" if d["active"] == 1 else "A0"
sl_part = ""
if d["stop_loss_pct"] > 0 and d["to_stoploss"] is not None:
sl_part = f" / to_SL={d['to_stoploss']:+.2f}%(-{int(d['stop_loss_pct']*100)}%)"
sign = "+" if d["profit"] >= 0 else ""
return (
f"- {d['name']}({d['code']})[{a}] to_stop={d['to_stop']:+.2f}%{sl_part}\n"
f" cur={fmt_money(d['cur'])} stop={fmt_money(d['stop'])} eff_trailing={int(d['eff_trailing']*100)}% "
f"pnl={sign}{fmt_money(d['profit'])}원 ({d['rate']:+.2f}%)"
) risk_box = (
f"[리스크 레벨: {risk['level']}]\n"
f"- Trigger: {risk['trigger_cnt']} / Near: {risk['near_cnt']}\n"
f"- 위험노출: {fmt_money(risk['risk_exposure'])}원 ({risk['exposure_pct']:.1f}%)\n"
) msg = (
f"[16:00 리포트/기준선+손절+리스크]\n"
f"- 총 손익(active=1): {fmt_money(total_profit_now)}원\n"
f"- 총 수익률(active=1): {total_rate_now:+.2f}%\n"
f"- 설명: to_stop=+X%는 트레일 기준선까지 남은 비율(0% 이하 트리거)\n"
f"- 설명: to_SL=+X%는 손절선까지 남은 비율(0% 이하 손절)\n\n"
+ risk_box + "\n"
+ "🚨 TRIGGER\n" + ("\n".join(line(d) for d in trigger) if trigger else "- 없음") + "\n\n"
+ f"⚠️ NEAR (<= {near_dist:.1f}%)\n" + ("\n".join(line(d) for d in near) if near else "- 없음") + "\n\n"
+ "✅ SAFE\n" + ("\n".join(line(d) for d in safe) if safe else "- 없음") +
f"\n\n- 시각: {now_str}"
)
send_synology_chat(webhook_url, msg)
state.setdefault("_after_close_summary", {})["last_date"] = now.strftime("%Y-%m-%d") if changed_positions:
save_positions_csv_atomic(POSITIONS_PATH, positions_all) return changed_positions, risk["level"], market_opendef main():
cfg = load_json_safe(CONFIG_PATH, {})
if not cfg:
raise RuntimeError("config.json이 없습니다.") default_trailing_pct = float(cfg.get("default_trailing_pct", 0.10))
default_stop_loss_pct = float(cfg.get("default_stop_loss_pct", 0.08)) state = load_json_safe(STATE_PATH, {}) if bool(cfg.get("startup_notify_enabled", True)) and should_send_startup_notify(state):
try:
positions_all = load_positions_csv(POSITIONS_PATH, default_trailing_pct, default_stop_loss_pct)
active_positions = [p for p in positions_all if p.active == 1]
lines = [
f"- {p.name}({p.code}.{p.market}) qty={p.qty:g} buy={fmt_money(p.buy_price)} "
f"trail={int(p.trailing_pct*100)}% sl={int(p.stop_loss_pct*100)}%"
for p in active_positions
]
msg = "[주식 알림봇 시작]\n" + "\n".join(lines) + f"\n- 시각: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
send_synology_chat(cfg["synology_chat_webhook_url"], msg)
mark_startup_notify_sent(state)
save_json_atomic(STATE_PATH, state)
except Exception as e:
log("STARTUP", f"시작 안내 실패: {e}", level="WARN") while True:
now = datetime.now()
try:
positions_all = load_positions_csv(POSITIONS_PATH, default_trailing_pct, default_stop_loss_pct) if bool(cfg.get("daily_reactivate_enabled", True)):
if should_reactivate_today(state, now, cfg.get("daily_reactivate_time", "09:01")):
if reactivate_all_positions(positions_all, state):
save_positions_csv_atomic(POSITIONS_PATH, positions_all)
save_json_atomic(STATE_PATH, state) changed, risk_level, market_open = run_once(cfg, positions_all, state)
save_json_atomic(STATE_PATH, state) sleep_min = int(cfg.get("interval_minutes_market", 30))
if bool(cfg.get("adaptive_interval_enabled", True)):
mkt_min = int(cfg.get("interval_minutes_market", sleep_min))
high_min = int(cfg.get("interval_minutes_high_risk", 10))
off_min = int(cfg.get("interval_minutes_off_market", 120))
sleep_min = (high_min if risk_level == "HIGH" else mkt_min) if market_open else off_min log("SLEEP", f"{sleep_min}분 후 재실행 (risk={risk_level}, market_open={market_open})")
time.sleep(sleep_min * 60) except Exception as e:
log("ERROR", f"루프 오류: {e}", level="ERROR")
time.sleep(120)if __name__ == "__main__":
main()
8) 실행(백그라운드)
cd ~/python/jusik_top10
nohup python3.9 main.py > jusik.log 2>&1 &
tail -f jusik.log
9) 운영 팁
positions.csv에서active=0으로 수동 비활성화 가능- 손절/트레일링 트리거 후
auto_disable_on_alert=true면 자동으로 active=0 - 다음날
09:01에 자동으로 active=1 복구 (옵션)
10) 트러블슈팅
✅ config.json이 없습니다
main.py와 같은 폴더에config.json이 있는지 확인
ls -al
✅ state.json JSONDecodeError(Extra data)
- 파일이 깨진 것 →
{}로 초기화
cp state.json state.json.bak
echo "{}" > state.json
✅ 알림이 안 오면
- webhook URL 오타/토큰 확인
- 방화벽/SSL 문제 여부 확인
- logs 확인
tail -n 200 jusik.log
이 글이 도움이 되었나요?좋아요/추천은 다시 누르면 취소됩니다.
