"실시간_설정_감시_Watchdog_기능_구현"
This commit is contained in:
@@ -6,6 +6,7 @@ from pydantic import BaseModel
|
|||||||
from app.db.database import get_db
|
from app.db.database import get_db
|
||||||
from app.db.models import ApiSettings
|
from app.db.models import ApiSettings
|
||||||
from app.services.kis_auth import kis_auth
|
from app.services.kis_auth import kis_auth
|
||||||
|
from app.workers.scheduler import service_watchdog_job
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
router = APIRouter()
|
router = APIRouter()
|
||||||
@@ -97,6 +98,9 @@ async def update_settings(payload: SettingsSchema, db: AsyncSession = Depends(ge
|
|||||||
await db.commit()
|
await db.commit()
|
||||||
await db.refresh(settings)
|
await db.refresh(settings)
|
||||||
|
|
||||||
|
# Trigger Startup-like sequence immediately
|
||||||
|
await service_watchdog_job()
|
||||||
|
|
||||||
# Return masked object check
|
# Return masked object check
|
||||||
# We return what was saved, but masked?
|
# We return what was saved, but masked?
|
||||||
# Usually convention is to return updated state.
|
# Usually convention is to return updated state.
|
||||||
|
|||||||
@@ -10,14 +10,40 @@ from app.services.telegram_service import telegram_service
|
|||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
async def check_kis_connectivity(db_session, settings_obj):
|
||||||
|
if not settings_obj.appKey or not settings_obj.appSecret:
|
||||||
|
logger.warning(">> [KIS] Credentials NOT FOUND in DB.")
|
||||||
|
return False
|
||||||
|
|
||||||
|
logger.info(">> [KIS] Attempting Authentication...")
|
||||||
|
try:
|
||||||
|
token = await kis_auth.get_access_token(db_session)
|
||||||
|
masked_token = token[:10] + "..." if token else "None"
|
||||||
|
logger.info(f" [OK] Access Token Valid ({masked_token})")
|
||||||
|
return True
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f" [FAILED] KIS Auth Failed: {e}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
async def check_telegram_connectivity(settings_obj, is_startup=False):
|
||||||
|
if settings_obj.useTelegram and settings_obj.telegramToken and settings_obj.telegramChatId:
|
||||||
|
logger.info(">> [Telegram] Integration Enabled.")
|
||||||
|
if is_startup:
|
||||||
|
msg = "🚀 <b>BatchuKis 배취키스</b> 시스템이 시작되었습니다.\n자동매매 엔진이 가동 중입니다."
|
||||||
|
await telegram_service.send_message(settings_obj.telegramToken, settings_obj.telegramChatId, msg)
|
||||||
|
return True
|
||||||
|
else:
|
||||||
|
# logger.debug(">> [Telegram] Disabled or missing config.")
|
||||||
|
return False
|
||||||
|
|
||||||
async def run_startup_sequence():
|
async def run_startup_sequence():
|
||||||
"""
|
"""
|
||||||
Executes the Phase 1~4 startup sequence defined in ReadMe.md.
|
Executes the Phase 1~4 startup sequence.
|
||||||
"""
|
"""
|
||||||
logger.info("=== Starting System Initialization Sequence ===")
|
logger.info("=== Starting System Initialization Sequence ===")
|
||||||
|
|
||||||
async with SessionLocal() as db_session:
|
async with SessionLocal() as db_session:
|
||||||
# Phase 1: DB & Settings Load
|
# Phase 1: Load Settings
|
||||||
stmt = select(ApiSettings).where(ApiSettings.id == 1)
|
stmt = select(ApiSettings).where(ApiSettings.id == 1)
|
||||||
result = await db_session.execute(stmt)
|
result = await db_session.execute(stmt)
|
||||||
settings_obj = result.scalar_one_or_none()
|
settings_obj = result.scalar_one_or_none()
|
||||||
@@ -28,48 +54,15 @@ async def run_startup_sequence():
|
|||||||
await db_session.commit()
|
await db_session.commit()
|
||||||
logger.info("Created Default ApiSettings.")
|
logger.info("Created Default ApiSettings.")
|
||||||
|
|
||||||
# Phase 2: KIS Connectivity
|
# Phase 2: KIS & Telegram
|
||||||
if not settings_obj.appKey or not settings_obj.appSecret:
|
await check_kis_connectivity(db_session, settings_obj)
|
||||||
logger.warning(">> [Phase 2] KIS Credentials (appKey/Secret) NOT FOUND in DB.")
|
await check_telegram_connectivity(settings_obj, is_startup=True)
|
||||||
logger.warning(" Please configure them via the Settings Page.")
|
|
||||||
logger.warning(" Skipping Token Issue & Realtime Connection.")
|
|
||||||
else:
|
|
||||||
logger.info(">> [Phase 2] KIS Credentials Found. Attempting Authentication...")
|
|
||||||
try:
|
|
||||||
# 1. Access Token
|
|
||||||
token = await kis_auth.get_access_token(db_session)
|
|
||||||
masked_token = token[:10] + "..." if token else "None"
|
|
||||||
logger.info(f" [OK] Access Token Valid (Starts with: {masked_token})")
|
|
||||||
|
|
||||||
# 2. Approval Key (Optional, lazy load usually, but good to check)
|
# Phase 3: Data Sync
|
||||||
# approval_key = await kis_auth.get_approval_key(db_session)
|
logger.info(">> [Phase 3] Syncing Account & Master Data...")
|
||||||
# logger.info(" [OK] WebSocket Approval Key Issued.")
|
|
||||||
|
|
||||||
except Exception as e:
|
|
||||||
logger.error(f" [FAILED] Authentication Failed: {e}")
|
|
||||||
logger.error(" Please check your AppKey/Secret and ensure KIS API Server is reachable.")
|
|
||||||
|
|
||||||
# Phase 2.5: Telegram Integration
|
|
||||||
if settings_obj.useTelegram and settings_obj.telegramToken and settings_obj.telegramChatId:
|
|
||||||
logger.info(">> [Phase 2.5] Telegram Integration Enabled. Sending Startup Notification...")
|
|
||||||
msg = "🚀 <b>BatchuKis 배취키스</b> 시스템이 시작되었습니다.\n자동매매 엔진이 가동 중입니다."
|
|
||||||
await telegram_service.send_message(
|
|
||||||
settings_obj.telegramToken,
|
|
||||||
settings_obj.telegramChatId,
|
|
||||||
msg
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
logger.info(">> [Phase 2.5] Telegram Disabled or Token/ChatID missing.")
|
|
||||||
|
|
||||||
# Phase 3: Data Sync (Master Stocks & Account)
|
|
||||||
logger.info(">> [Phase 3-1] Syncing Account Data...")
|
|
||||||
await sync_service.sync_account(db_session)
|
await sync_service.sync_account(db_session)
|
||||||
|
|
||||||
logger.info(">> [Phase 3-2] Syncing Master Data (This may take a while)...")
|
|
||||||
from app.services.master_service import master_service
|
from app.services.master_service import master_service
|
||||||
await master_service.sync_master_data(db_session)
|
await master_service.sync_master_data(db_session)
|
||||||
|
|
||||||
# Phase 4: Scheduler
|
|
||||||
# (Scheduler is started in main.py)
|
|
||||||
|
|
||||||
logger.info("=== System Initialization Complete ===")
|
logger.info("=== System Initialization Complete ===")
|
||||||
|
|||||||
@@ -3,15 +3,60 @@ from app.core.market_schedule import market_schedule
|
|||||||
from app.services.kis_client import kis_client
|
from app.services.kis_client import kis_client
|
||||||
from app.services.realtime_manager import realtime_manager
|
from app.services.realtime_manager import realtime_manager
|
||||||
from app.db.database import SessionLocal
|
from app.db.database import SessionLocal
|
||||||
from app.db.models import StockItem, ReservedOrder
|
from app.core.startup import check_kis_connectivity, check_telegram_connectivity
|
||||||
from sqlalchemy import select, update
|
from app.db.models import StockItem, ReservedOrder, ApiSettings
|
||||||
import logging
|
|
||||||
from datetime import datetime
|
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
scheduler = AsyncIOScheduler()
|
scheduler = AsyncIOScheduler()
|
||||||
|
|
||||||
|
class ServiceState:
|
||||||
|
last_keys_hash: str = ""
|
||||||
|
telegram_active: bool = False
|
||||||
|
naver_active: bool = False
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def get_keys_hash(cls, settings: ApiSettings):
|
||||||
|
return f"{settings.appKey}{settings.appSecret}{settings.accountNumber}"
|
||||||
|
|
||||||
|
async def service_watchdog_job():
|
||||||
|
"""
|
||||||
|
Periodically checks settings and initiates services if they were newly enabled or changed.
|
||||||
|
"""
|
||||||
|
async with SessionLocal() as db_session:
|
||||||
|
stmt = select(ApiSettings).where(ApiSettings.id == 1)
|
||||||
|
result = await db_session.execute(stmt)
|
||||||
|
settings_obj = result.scalar_one_or_none()
|
||||||
|
|
||||||
|
if not settings_obj: return
|
||||||
|
|
||||||
|
# 1. KIS Connectivity Watchdog
|
||||||
|
current_hash = ServiceState.get_keys_hash(settings_obj)
|
||||||
|
if current_hash != ServiceState.last_keys_hash and settings_obj.appKey:
|
||||||
|
logger.info("Watchdog: KIS Credentials Changed or Initialized. Re-authenticating...")
|
||||||
|
success = await check_kis_connectivity(db_session, settings_obj)
|
||||||
|
if success:
|
||||||
|
ServiceState.last_keys_hash = current_hash
|
||||||
|
|
||||||
|
# 2. Telegram Watchdog
|
||||||
|
if settings_obj.useTelegram and not ServiceState.telegram_active:
|
||||||
|
if settings_obj.telegramToken and settings_obj.telegramChatId:
|
||||||
|
logger.info("Watchdog: Telegram newly enabled. Sending activation message.")
|
||||||
|
from app.services.telegram_service import telegram_service
|
||||||
|
msg = "🔔 <b>알림 서비스가 활성화되었습니다.</b>\n이제부터 주요 거래 알림을 이곳으로 보내드립니다."
|
||||||
|
await telegram_service.send_message(settings_obj.telegramToken, settings_obj.telegramChatId, msg)
|
||||||
|
ServiceState.telegram_active = True
|
||||||
|
elif not settings_obj.useTelegram:
|
||||||
|
ServiceState.telegram_active = False
|
||||||
|
|
||||||
|
# 3. Naver News Watchdog
|
||||||
|
if settings_obj.useNaverNews and not ServiceState.naver_active:
|
||||||
|
if settings_obj.naverClientId and settings_obj.naverClientSecret:
|
||||||
|
logger.info("Watchdog: Naver News newly enabled. (Logic placeholder)")
|
||||||
|
ServiceState.naver_active = True
|
||||||
|
elif not settings_obj.useNaverNews:
|
||||||
|
ServiceState.naver_active = False
|
||||||
|
|
||||||
async def market_check_job():
|
async def market_check_job():
|
||||||
"""
|
"""
|
||||||
Periodic check to ensure Realtime Manager is connected when market is open.
|
Periodic check to ensure Realtime Manager is connected when market is open.
|
||||||
@@ -167,6 +212,7 @@ def start_scheduler():
|
|||||||
scheduler.add_job(market_check_job, 'interval', minutes=5)
|
scheduler.add_job(market_check_job, 'interval', minutes=5)
|
||||||
scheduler.add_job(news_scrap_job, 'interval', minutes=10)
|
scheduler.add_job(news_scrap_job, 'interval', minutes=10)
|
||||||
scheduler.add_job(auto_trade_scan_job, 'interval', minutes=1)
|
scheduler.add_job(auto_trade_scan_job, 'interval', minutes=1)
|
||||||
|
scheduler.add_job(service_watchdog_job, 'interval', minutes=1) # Check settings every 1 min
|
||||||
scheduler.add_job(persist_market_data_job, 'interval', seconds=10) # Persist every 10s
|
scheduler.add_job(persist_market_data_job, 'interval', seconds=10) # Persist every 10s
|
||||||
|
|
||||||
scheduler.start()
|
scheduler.start()
|
||||||
|
|||||||
Reference in New Issue
Block a user