import os
import requests
from datetime import datetime
from pykrx import stock
import pandas as pd
import ta
from dotenv import load_dotenv
# ============================================================
# DC 퇴직연금 100거래일 매일 분할매수
# ============================================================
# 총 투자금 : 250,000,000원
# 분할매수 기간 : 100거래일
# 하루 총 투자금 : 2,500,000원
# 위험자산 70% : 1,750,000원
# 안전자산 30% : 750,000원
#
# RSI는 위험자산 175만원 안에서만 동적 배분
# → 하루 위험자산 총액은 절대 증가하지 않음
#
# 실제 주문은 하지 않고 Telegram으로 수량만 안내
# ============================================================
load_dotenv()
TOKEN = os.getenv(“TELEGRAM_TOKEN”)
CHAT_ID = os.getenv(“CHAT_ID”)
# ============================================================
# 기본 설정
# ============================================================
TOTAL_CAPITAL = 250_000_000
FIXED_TRADING_DAYS = 100
RISK_RATIO = 0.70
SAFE_RATIO = 0.30
# 하루 투자금
DAILY_TOTAL_BUDGET = (
TOTAL_CAPITAL / FIXED_TRADING_DAYS
)
# 하루 위험자산
DAILY_RISK_BUDGET = (
DAILY_TOTAL_BUDGET * RISK_RATIO
)
# 하루 안전자산
DAILY_SAFE_BUDGET = (
DAILY_TOTAL_BUDGET * SAFE_RATIO
)
# ============================================================
# 투자기간
# ============================================================
START_DATE = “20260730”
END_DATE = “20270131”
# ============================================================
# 위험자산 포트폴리오
#
# 전체 자산 기준:
# 나스닥 25%
# S&P500 20%
# TIGER200 15%
# 반도체 10%
#
# 합계 70%
#
# 아래 weight는 위험자산 내부 비중으로 사용되며
# 자동으로 100% 정규화됩니다.
# ============================================================
TARGET_PORTFOLIO = {
“133690”: {
“name”: “TIGER 미국나스닥100”,
“weight”: 0.25,
},
“360750”: {
“name”: “TIGER 미국S&P500”,
“weight”: 0.20,
},
“102110”: {
“name”: “TIGER 200”,
“weight”: 0.15,
},
“396500”: {
“name”: “KODEX 반도체TOP10”,
“weight”: 0.10,
},
}
# ============================================================
# RSI 설정
# ============================================================
#
# RSI가 낮으면 해당 ETF 비중 증가
# RSI가 높으면 해당 ETF 비중 감소
#
# 중요:
# RSI 배수 적용 후 다시 100%로 정규화하므로
# 하루 위험자산 총액 175만원은 그대로 유지됩니다.
# ============================================================
def get_rsi_multiplier(rsi):
if rsi < 25:
return 1.20, “🔴🔴 매우 과매도”
elif rsi < 35:
return 1.10, “🔴 과매도”
elif rsi <= 65:
return 1.00, “⚪ 보통”
elif rsi <= 75:
return 0.90, “🔵 과열”
else:
return 0.80, “🔵🔵 매우 과열”
# ============================================================
# Telegram
# ============================================================
def send_telegram(msg):
if not TOKEN or not CHAT_ID:
print(“⚠️ TELEGRAM_TOKEN 또는 CHAT_ID가 없습니다.”)
print(msg)
return
url = (
f”https://api.telegram.org/”
f”bot{TOKEN}/sendMessage”
)
try:
response = requests.post(
url,
data={
“chat_id”: CHAT_ID,
“text”: msg,
“parse_mode”: “Markdown”,
},
timeout=10,
)
response.raise_for_status()
except Exception as e:
print(
f”Telegram 전송 오류: {e}”
)
print(msg)
# ============================================================
# 오늘 실제 거래일 확인
# ============================================================
def is_today_trading_day():
today = datetime.today().strftime(
“%Y%m%d”
)
df = stock.get_market_ohlcv_by_date(
today,
today,
“005930”,
)
return not df.empty
# ============================================================
# RSI / 현재가 분석
# ============================================================
def get_stock_analysis(ticker):
today = datetime.today().strftime(
“%Y%m%d”
)
df = stock.get_market_ohlcv_by_date(
“20260101”,
today,
ticker,
)
if df.empty or len(df) < 20:
return None
rsi_series = ta.momentum.rsi(
df[“종가”],
window=14,
)
rsi = rsi_series.iloc[-1]
current_price = int(
df[“종가”].iloc[-1]
)
if pd.isna(rsi):
return None
multiplier, status = (
get_rsi_multiplier(
float(rsi)
)
)
return {
“price”: current_price,
“rsi”: round(
float(rsi),
2,
),
“multiplier”: multiplier,
“status”: status,
}
# ============================================================
# 오늘 매수 계산
# ============================================================
def calculate_daily_orders():
analyses = []
# ——————————————————–
# 각 ETF 분석
# ——————————————————–
for ticker, info in TARGET_PORTFOLIO.items():
analysis = get_stock_analysis(
ticker
)
if not analysis:
continue
adjusted_weight = (
info[“weight”]
* analysis[“multiplier”]
)
analyses.append({
“ticker”: ticker,
“info”: info,
“analysis”: analysis,
“adjusted_weight”:
adjusted_weight,
})
# ——————————————————–
# 분석 실패
# ——————————————————–
if not analyses:
return [], 0
# ——————————————————–
# RSI 적용 후 총 가중치
# ——————————————————–
total_adjusted_weight = sum(
item[“adjusted_weight”]
for item in analyses
)
orders = []
total_cost = 0
# ——————————————————–
# 종목별 주문 계산
# ——————————————————–
for item in analyses:
ticker = item[“ticker”]
info = item[“info”]
analysis = item[“analysis”]
# —————————————————-
# RSI 적용 후 위험자산 내부 비중
# —————————————————-
normalized_weight = (
item[“adjusted_weight”]
/ total_adjusted_weight
)
# —————————————————-
# 위험자산 175만원 안에서 배분
# —————————————————-
target_amount = (
DAILY_RISK_BUDGET
* normalized_weight
)
price = analysis[“price”]
# —————————————————-
# 1주 단위 매수
# —————————————————-
quantity = int(
target_amount
// price
)
cost = (
quantity
* price
)
total_cost += cost
orders.append({
“ticker”: ticker,
“name”: info[“name”],
“price”: price,
“rsi”: analysis[“rsi”],
“multiplier”:
analysis[“multiplier”],
“status”:
analysis[“status”],
“weight”:
normalized_weight,
“target_amount”:
target_amount,
“quantity”:
quantity,
“cost”:
cost,
})
return orders, total_cost
# ============================================================
# Telegram 리포트
# ============================================================
def build_report(
today_str,
orders,
total_risk_cost,
):
today_display = datetime.strptime(
today_str,
“%Y%m%d”,
).strftime(
“%Y-%m-%d”
)
# ——————————————————–
# 기본 정보
# ——————————————————–
msg = “”
msg += (
f”📊 **[{today_display}] “
f”DC 퇴직연금 100거래일 분할매수**\n”
)
msg += (
“━━━━━━━━━━━━━━━━━━━━\n”
)
msg += (
f”🎯 총 투자금: “
f”**{TOTAL_CAPITAL:,.0f}원**\n”
f”🗓️ 분할매수: “
f”**{FIXED_TRADING_DAYS}거래일**\n”
f”📅 오늘 총 투자예산: “
f”**{DAILY_TOTAL_BUDGET:,.0f}원**\n\n”
f”📈 위험자산 70%: “
f”**{DAILY_RISK_BUDGET:,.0f}원**\n”
f”💵 안전자산 30%: “
f”**{DAILY_SAFE_BUDGET:,.0f}원**\n”
)
msg += (
“━━━━━━━━━━━━━━━━━━━━\n”
)
# ——————————————————–
# ETF 주문
# ——————————————————–
for item in orders:
msg += (
f”\n📌 **{item[‘name’]}** “
f”({item[‘ticker’]})\n”
)
msg += (
f”• 현재가: “
f”{item[‘price’]:,}원\n”
f”• RSI: “
f”{item[‘rsi’]} “
f”{item[‘status’]}\n”
f”• RSI 배수: “
f”{item[‘multiplier’]:.2f}x\n”
f”• 위험자산 내 비중: “
f”**{item[‘weight’] * 100:.1f}%**\n”
f”• **매수: “
f”{item[‘quantity’]}주** “
f”({item[‘cost’]:,}원)\n”
)
# ——————————————————–
# 합계
# ——————————————————–
unused_risk = (
DAILY_RISK_BUDGET
– total_risk_cost
)
msg += (
“\n━━━━━━━━━━━━━━━━━━━━\n”
)
msg += (
f”🛒 위험자산 실제 매수: “
f”**{total_risk_cost:,.0f}원**\n”
f”📊 위험자산 미사용 잔액: “
f”**{unused_risk:,.0f}원**\n”
f”💵 안전자산 적립: “
f”**{DAILY_SAFE_BUDGET:,.0f}원**\n\n”
f”📌 하루 총 투자 한도: “
f”**{DAILY_TOTAL_BUDGET:,.0f}원**\n”
)
msg += (
“\n📢 *미래에셋 M-STOCK 앱에서 “
“수량대로 수동 주문하세요.*”
)
return msg
# ============================================================
# MAIN
# ============================================================
def main():
today = datetime.today().strftime(
“%Y%m%d”
)
# ——————————————————–
# 투자기간 확인
# ——————————————————–
if today < START_DATE or today > END_DATE:
print(
“현재 날짜가 투자기간 밖입니다.”
)
return
# ——————————————————–
# 실제 거래일 확인
# ——————————————————–
if not is_today_trading_day():
print(
“휴장일이므로 종료합니다.”
)
return
# ——————————————————–
# 주문 계산
# ——————————————————–
orders, total_risk_cost = (
calculate_daily_orders()
)
if not orders:
print(
“ETF 데이터를 가져오지 못했습니다.”
)
return
# ——————————————————–
# 안전장치
#
# 어떤 이유로 계산 오류가 발생해도
# 위험자산 175만원을 초과하지 않음
# ——————————————————–
if total_risk_cost > DAILY_RISK_BUDGET:
print(
“⚠️ 위험자산 매수금액이 “
“일일 한도를 초과했습니다.”
)
return
# ——————————————————–
# Telegram
# ——————————————————–
msg = build_report(
today_str=today,
orders=orders,
total_risk_cost=
total_risk_cost,
)
send_telegram(msg)
# ——————————————————–
# 콘솔 출력
# ——————————————————–
print(msg)
# ============================================================
# 실행
# ============================================================
if __name__ == “__main__”:
main()
Leave a Reply