"백엔드_핵심_로직_구현_프론트엔드_연동_및_도커_배포_최적화_완료"

backend/app/services/kis_auth.py

@@ -5,13 +5,14 @@ from app.db.database import SessionLocal
 from app.db.models import ApiSettings 
 from app.db.models import ApiSettings 
 from app.core.crypto import decrypt_str, encrypt_str
+import logging
 
 class KisAuth:
     BASE_URL_REAL = "https://openapi.koreainvestment.com:9443"
     # BASE_URL_VIRTUAL = "https://openapivts.koreainvestment.com:29443"
     
     def __init__(self):
-        pass
+        self.logger = logging.getLogger(self.__class__.__name__)
 
     async def get_access_token(self, db_session=None) -> str:
         """
@@ -37,9 +38,11 @@ class KisAuth:
                  token_dec = decrypt_str(settings_obj.accessToken)
                  if token_dec and token_dec != "[Decryption Failed]":
                      if settings_obj.tokenExpiry > datetime.now() + timedelta(minutes=10):
+                         # self.logger.debug("Using cached Access Token.") # Too verbose?
                          return token_dec
             
             # 3. Issue New Token
+            self.logger.info("Access Token Expired or Missing. Issuing New Token...")
             app_key_dec = decrypt_str(settings_obj.appKey)
             app_secret_dec = decrypt_str(settings_obj.appSecret)
             token_data = await self._issue_token(app_key_dec, app_secret_dec)

backend/app/services/kis_client.py

@@ -6,6 +6,7 @@ from app.db.database import SessionLocal
 from app.db.models import ApiSettings 
 from sqlalchemy import select
 from app.core.crypto import decrypt_str
+import logging
 
 class KisClient:
     """
@@ -15,16 +16,19 @@ class KisClient:
     
     # Domestic URLs
     URL_DOMESTIC_ORDER = "/uapi/domestic-stock/v1/trading/order-cash"
+    URL_DOMESTIC_MODIFY = "/uapi/domestic-stock/v1/trading/order-rvsecncl"
     URL_DOMESTIC_PRICE = "/uapi/domestic-stock/v1/quotations/inquire-price"
     URL_DOMESTIC_BALANCE = "/uapi/domestic-stock/v1/trading/inquire-balance"
     
     # Overseas URLs
     URL_OVERSEAS_ORDER = "/uapi/overseas-stock/v1/trading/order"
+    URL_OVERSEAS_MODIFY = "/uapi/overseas-stock/v1/trading/order-rvsecncl"
     URL_OVERSEAS_PRICE = "/uapi/overseas-price/v1/quotations/price"
     URL_OVERSEAS_BALANCE = "/uapi/overseas-stock/v1/trading/inquire-balance"
 
     def __init__(self):
         pass
+        self.logger = logging.getLogger(self.__class__.__name__)
 
     async def _get_settings(self):
         async with SessionLocal() as session:
@@ -60,6 +64,7 @@ class KisClient:
         }
         
         full_url = f"{base_url}{url_path}"
+        # self.logger.debug(f"API Calling: {method} {url_path} (TR_ID: {tr_id})")
         
         async with httpx.AsyncClient() as client:
             if method == "GET":
@@ -178,6 +183,7 @@ class KisClient:
                 "ORD_QTY": str(quantity),
                 "ORD_UNPR": str(int(price)), # Cash Order requires integer price string
             }
+            self.logger.info(f"Ordering Domestic: {side} {code} {quantity}qty @ {price}")
             return await self._call_api("POST", self.URL_DOMESTIC_ORDER, tr_id, data=data)
             
         elif market == "Overseas":
@@ -197,4 +203,52 @@ class KisClient:
              }
              return await self._call_api("POST", self.URL_OVERSEAS_ORDER, tr_id, data=data)
 
+    async def modify_order(self, market: str, order_no: str, code: str, quantity: int, price: float, type: str = "00", cancel: bool = False) -> Dict:
+        """
+        Cancel or Modify Order.
+        cancel=True -> Cancel
+        """
+        settings = await self._get_settings()
+        acc_no_str = decrypt_str(settings.accountNumber)
+        
+        if '-' in acc_no_str:
+            cano, prdt = acc_no_str.split('-')
+        else:
+            cano = acc_no_str[:8]
+            prdt = acc_no_str[8:]
+            
+        if market == "Domestic":
+            # TR_ID: TTTC0803U (Modify/Cancel)
+            data = {
+                "CANO": cano,
+                "ACNT_PRDT_CD": prdt,
+                "KRX_FWDG_ORD_ORGNO": "", # Exchange Node? Usually empty or "00950"
+                "ORGN_ODNO": order_no,
+                "ORD_DVSN": type, 
+                "RVSE_CNCL_DVSN_CD": "02" if cancel else "01", # 01: Modify, 02: Cancel
+                "ORD_QTY": str(quantity),
+                "ORD_UNPR": str(int(price)),
+                "QTY_ALL_ORD_YN": "Y" if quantity == 0 else "N", # 0 means cancel all?
+            }
+            # Note: KRX_FWDG_ORD_ORGNO is tricky. Usually 5 digit branch code. Defaulting to "" might fail.
+            # Using '06010' (Online) or leaving blank depending on API.
+            # KIS API Doc: "주문상태조회"에서 얻은 ORGNO 사용해야 함.
+            # For this impl, we assume user knows or simple default.
+            
+            return await self._call_api("POST", self.URL_DOMESTIC_MODIFY, "TTTC0803U", data=data)
+            
+        elif market == "Overseas":
+            # MCCL: TTTT1004U
+             data = {
+                "CANO": cano,
+                "ACNT_PRDT_CD": prdt,
+                "OVRS_EXCG_CD": "NASD", 
+                "PDNO": code,
+                "ORGN_ODNO": order_no,
+                "RVSE_CNCL_DVSN_CD": "02" if cancel else "01",
+                "ORD_QTY": str(quantity),
+                "OVRS_ORD_UNPR": str(price),
+             }
+             return await self._call_api("POST", self.URL_OVERSEAS_MODIFY, "TTTT1004U", data=data)
+
 kis_client = KisClient()

backend/app/services/master_service.py

@@ -0,0 +1,112 @@
+import os
+import zipfile
+import httpx
+import logging
+import asyncio
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import delete
+from app.db.models import MasterStock
+from app.db.database import SessionLocal
+
+logger = logging.getLogger(__name__)
+
+class MasterService:
+    BASE_URL = "https://new.real.download.dws.co.kr/common/master"
+    FILES = {
+        "KOSPI": "kospi_code.mst.zip",
+        "KOSDAQ": "kosdaq_code.mst.zip"
+    }
+    TMP_DIR = "./tmp_master"
+
+    async def sync_master_data(self, db: AsyncSession):
+        """
+        Download and parse KOSPI/KOSDAQ master files.
+        Populate MasterStock table.
+        """
+        logger.info("MasterService: Starting Master Data Sync...")
+        os.makedirs(self.TMP_DIR, exist_ok=True)
+        
+        try:
+            # Clear existing data? Or Upsert?
+            # For simplicity, Clear and Re-insert (Full Sync)
+            # await db.execute(delete(MasterStock)) # Optional: Clear all
+            
+            total_count = 0
+            
+            async with httpx.AsyncClient(verify=False) as client:
+                for market, filename in self.FILES.items():
+                    url = f"{self.BASE_URL}/{filename}"
+                    dest = os.path.join(self.TMP_DIR, filename)
+                    
+                    # 1. Download
+                    logger.info(f"Downloading {market} from {url}...")
+                    try:
+                        resp = await client.get(url, timeout=60.0)
+                        resp.raise_for_status()
+                        with open(dest, "wb") as f:
+                            f.write(resp.content)
+                    except Exception as e:
+                        logger.error(f"Failed to download {market}: {e}")
+                        continue
+
+                    # 2. Unzip & Parse
+                    count = await self._process_zip(dest, market, db)
+                    total_count += count
+            
+            await db.commit()
+            logger.info(f"MasterService: Sync Complete. Total {total_count} stocks.")
+            
+            # Cleanup
+            # shutil.rmtree(self.TMP_DIR) 
+
+        except Exception as e:
+            logger.error(f"MasterService: Fatal Error: {e}")
+            await db.rollback()
+
+    async def _process_zip(self, zip_path: str, market: str, db: AsyncSession) -> int:
+        try:
+            with zipfile.ZipFile(zip_path, 'r') as zf:
+                mst_filename = zf.namelist()[0] # Usually only one .mst file
+                zf.extract(mst_filename, self.TMP_DIR)
+                mst_path = os.path.join(self.TMP_DIR, mst_filename)
+                
+                return await self._parse_mst(mst_path, market, db)
+        except Exception as e:
+            logger.error(f"Error processing ZIP {zip_path}: {e}")
+            return 0
+
+    async def _parse_mst(self, mst_path: str, market: str, db: AsyncSession) -> int:
+        count = 0
+        batch = []
+        
+        # Encoding is usually cp949 for KIS files
+        with open(mst_path, "r", encoding="cp949", errors="replace") as f:
+            for line in f:
+                # Format:
+                # row[0:9] : Short Code (Example: "005930   ")
+                # row[9:21] : Standard Code
+                # row[21:len-222] : Name
+                
+                if len(line) < 250: continue # Invalid line
+                
+                short_code = line[0:9].strip()
+                # standard_code = line[9:21].strip()
+                name_part = line[21:len(line)-222].strip()
+                
+                if not short_code or not name_part: continue
+                
+                # Check for ETF/ETN? (Usually included)
+                
+                obj = MasterStock(
+                    code=short_code,
+                    name=name_part,
+                    market=market
+                )
+                db.add(obj)
+                count += 1
+                
+                # Batch commit? session.add is fast, commit at end.
+                
+        return count
+
+master_service = MasterService()

backend/app/services/realtime_manager.py

@@ -3,6 +3,7 @@ import json
 import websockets
 import logging
 from typing import Dict, Set, Callable, Optional
+from datetime import datetime
 
 from app.services.kis_auth import kis_auth
 from app.core.crypto import aes_cbc_base64_dec
@@ -21,134 +22,113 @@ class RealtimeManager:
     def __init__(self):
         self.ws: Optional[websockets.WebSocketClientProtocol] = None
         self.approval_key: Optional[str] = None
-        self.subscribed_codes: Set[str] = set()
-        self.running = False
-        self.data_map: Dict[str, Dict] = {} # Store IV/Key for encrypted TRs
-
-    async def start(self):
-        """
-        Main loop: Connect -> Authenticate -> Listen
-        """
-        self.running = True
-        while self.running:
-            try:
-                # 1. Get Approval Key
-                self.approval_key = await kis_auth.get_approval_key()
-                
-                logger.info(f"Connecting to KIS WS: {self.WS_URL_REAL}")
-                async with websockets.connect(self.WS_URL_REAL, ping_interval=None) as websocket:
-                    self.ws = websocket
-                    logger.info("Connected.")
-                    
-                    # 2. Resubscribe if recovering connection
-                    if self.subscribed_codes:
-                        await self._resubscribe_all()
-
-                    # 3. Listen Loop
-                    await self._listen()
-                    
-            except Exception as e:
-                logger.error(f"WS Connection Error: {e}. Retrying in 5s...")
-                await asyncio.sleep(5)
-
-    async def stop(self):
-        self.running = False
-        if self.ws:
-            await self.ws.close()
-
-    async def subscribe(self, stock_code: str, type="price"):
-        """
-        Subscribe to a stock.
-        type: 'price' (H0STCNT0 - 체결가)
-        """
-        if not self.ws or not self.approval_key:
-            logger.warning("WS not ready. Adding to pending list.")
-            self.subscribed_codes.add(stock_code)
-            return
-
-        # Domestic Realtime Price TR ID: H0STCNT0
-        tr_id = "H0STCNT0" 
-        tr_key = stock_code
         
+        # Reference Counting: Code -> Set of Sources
+        # e.g. "005930": {"HOLDING", "FRONTEND_DASHBOARD"}
+        self.subscriptions: Dict[str, Set[str]] = {} 
+        
+        self.running = False
+        self.data_map: Dict[str, Dict] = {} 
+        
+        # Realtime Data Cache (Code -> DataDict)
+        # Used by Scheduler to persist data periodically
+        self.price_cache: Dict[str, Dict] = {} 
+
+    async def add_subscription(self, code: str, source: str):
+        """
+        Request subscription. Increments reference count for the code.
+        """
+        if code not in self.subscriptions:
+            self.subscriptions[code] = set()
+            
+        if not self.subscriptions[code]:
+            # First subscriber, Send WS Command
+            await self._send_subscribe(code, "1") # 1=Register
+            
+        self.subscriptions[code].add(source)
+        logger.info(f"Subscribed {code} by {source}. RefCount: {len(self.subscriptions[code])}")
+
+    async def remove_subscription(self, code: str, source: str):
+        """
+        Remove subscription. Decrements reference count.
+        """
+        if code in self.subscriptions and source in self.subscriptions[code]:
+            self.subscriptions[code].remove(source)
+            logger.info(f"Unsubscribed {code} by {source}. RefCount: {len(self.subscriptions[code])}")
+            
+            if not self.subscriptions[code]:
+                # No more subscribers, Send WS Unsubscribe
+                await self._send_subscribe(code, "2") # 2=Unregister
+                del self.subscriptions[code]
+
+    async def _send_subscribe(self, code: str, tr_type: str):
+        if not self.ws or not self.approval_key:
+            return # Will resubscribe on connect
+            
         payload = {
             "header": {
                 "approval_key": self.approval_key,
                 "custtype": "P",
-                "tr_type": "1", # 1=Register, 2=Unregister
+                "tr_type": tr_type,
                 "content-type": "utf-8"
             },
             "body": {
                 "input": {
-                    "tr_id": tr_id,
-                    "tr_key": tr_key
+                    "tr_id": "H0STCNT0",
+                    "tr_key": code
                 }
             }
         }
-        
         await self.ws.send(json.dumps(payload))
-        self.subscribed_codes.add(stock_code)
-        logger.info(f"Subscribed to {stock_code}")
 
     async def _resubscribe_all(self):
-        for code in self.subscribed_codes:
-            await self.subscribe(code)
+        for code in list(self.subscriptions.keys()):
+            await self._send_subscribe(code, "1")
 
     async def _listen(self):
         async for message in self.ws:
             try:
-                # Message can be plain text provided by library, or bytes
-                if isinstance(message, bytes):
-                    message = message.decode('utf-8')
-
-                # KIS sends data in specific formats.
-                # 1. JSON (Control Messages, PINGPONG, Subscription Ack)
-                # 2. Text/Pipe separated (Real Data) - Usually starts with 0 or 1
+                if isinstance(message, bytes): message = message.decode('utf-8')
                 
                 first_char = message[0]
-                
-                if first_char in ['{', '[']: 
-                    # JSON Message
-                    data = json.loads(message)
-                    header = data.get('header', {})
-                    tr_id = header.get('tr_id')
-
-                    if tr_id == "PINGPONG":
-                        await self.ws.send(message) # Echo back
-                        logger.debug("PINGPONG handled")
-                        
-                    elif 'body' in data:
-                        # Subscription Ack
-                        # Store IV/Key if encryption is enabled (msg1 often contains 'ENCRYPT')
-                        # But for Brokerage API, H0STCNT0 is usually plaintext unless configured otherwise.
-                        # If encrypted, 'iv' and 'key' are in body['output']
-                        pass
-                        
-                elif first_char in ['0', '1']:
-                    # Real Data: 0|TR_ID|DATA_CNT|DATA...
+                if first_char in ['0', '1']:
+                    # Real Data
                     parts = message.split('|')
-                    if len(parts) < 4:
-                        continue
-                        
+                    if len(parts) < 4: continue
                     tr_id = parts[1]
                     raw_data = parts[3]
                     
-                    # Decryption Check
-                    # If this tr_id was registered as encrypted, decrypt it.
-                    # For now assuming Plaintext for H0STCNT0 as per standard Personal API.
-                    
-                    # Parse Data
-                    if tr_id == "H0STCNT0": # Domestic Price
-                        # Data format: TIME^PRICE^...
-                        # We need to look up format spec.
-                        # Simple implementation: just log or split
-                        fields = raw_data.split('^')
-                        if len(fields) > 2:
-                            current_price = fields[2] # Example index
-                            # TODO: Update DB
-                            # print(f"Price Update: {current_price}")
-                            pass
-                            
+                    if tr_id == "H0STCNT0":
+                        await self._parse_domestic_price(raw_data)
+                        
+                elif first_char == '{':
+                    data = json.loads(message)
+                    if data.get('header', {}).get('tr_id') == "PINGPONG":
+                        await self.ws.send(message)
+                        
             except Exception as e:
-                logger.error(f"Error processing WS message: {e}")
+                logger.error(f"WS Error: {e}")
 
+    async def _parse_domestic_price(self, raw_data: str):
+        # Format: MKSC_SHRN_ISCD^EXEC_TIME^CURRENT_PRICE^...
+        fields = raw_data.split('^')
+        if len(fields) < 3: return
+        
+        code = fields[0]
+        curr_price = fields[2]
+        change = fields[4]
+        change_rate = fields[5]
+        
+        # Create lightweight update object (Dict)
+        update_data = {
+            "code": code,
+            "price": curr_price,
+            "change": change,
+            "rate": change_rate,
+            "timestamp": datetime.now().isoformat()
+        }
+        
+        # Update Cache
+        self.price_cache[code] = update_data
+        # logger.debug(f"Price Update: {code} {curr_price}")
 realtime_manager = RealtimeManager()

backend/app/services/sync_service.py

@@ -0,0 +1,104 @@
+import logging
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, delete
+from datetime import datetime
+
+from app.db.database import SessionLocal
+from app.db.models import AccountStatus, Holding
+from app.services.kis_client import kis_client
+
+logger = logging.getLogger(__name__)
+
+class SyncService:
+    async def sync_account(self, db: AsyncSession):
+        """
+        Fetches balance from KIS and updates DB (AccountStatus & Holdings).
+        Currently supports Domestic only.
+        """
+        logger.info("SyncService: Starting Account Sync...")
+        try:
+            # 1. Fetch Domestic Balance
+            # kis_client.get_balance returns dict with 'output1', 'output2' or None
+            res = await kis_client.get_balance("Domestic")
+            if not res:
+                logger.error("SyncService: Failed to fetch balance (API Error or No Data).")
+                return
+
+            # output1: Holdings List
+            # output2: Account Summary
+            output1 = res.get('output1', [])
+            output2 = res.get('output2', [])
+            
+            # KIS API returns output2 as a LIST of 1 dict usually
+            summary_data = output2[0] if output2 else {}
+            
+            # --- Update AccountStatus ---
+            # Map KIS fields to AccountStatus model
+            # tot_evlu_amt: 총평가금액 (Total Assets)
+            # dnca_tot_amt: 예수금총액 (Buying Power)
+            # evlu_pfls_smt_tl: 평가손익합계 (Daily Profit - approximation)
+            # evlu_pfls_rt: 수익률
+            
+            total_assets = float(summary_data.get('tot_evlu_amt', 0) or 0)
+            buying_power = float(summary_data.get('dnca_tot_amt', 0) or 0)
+            daily_profit = float(summary_data.get('evlu_pfls_smt_tl', 0) or 0)
+            
+            # Calculate daily profit rate if not provided directly
+            # profit_rate = float(summary_data.get('evlu_pfls_rt', 0)) # Sometimes available
+            daily_profit_rate = 0.0
+            if total_assets > 0:
+                daily_profit_rate = (daily_profit / total_assets) * 100
+
+            # Upsert AccountStatus (ID=1)
+            stmt = select(AccountStatus).where(AccountStatus.id == 1)
+            result = await db.execute(stmt)
+            status = result.scalar_one_or_none()
+            
+            if not status:
+                status = AccountStatus(id=1)
+                db.add(status)
+            
+            status.totalAssets = total_assets
+            status.buyingPower = buying_power
+            status.dailyProfit = daily_profit
+            status.dailyProfitRate = daily_profit_rate
+            
+            # --- Update Holdings ---
+            # Strategy: Delete all existing holdings (refresh) or Upsert?
+            # Refresh is safer to remove sold items.
+            await db.execute(delete(Holding))
+            
+            for item in output1:
+                # Map fields
+                # pdno: 종목번호
+                # prdt_name: 종목명
+                # hldg_qty: 보유수량
+                # pchs_avg_pric: 매입평균가격
+                # prpr: 현재가
+                # evlu_pfls_amt: 평가손익금액
+                # evlu_pfls_rt: 평가손익율
+                # evlu_amt: 평가금액
+                
+                code = item.get('pdno')
+                if not code: continue
+                
+                h = Holding(
+                    stockCode=code,
+                    stockName=item.get('prdt_name', 'Unknown'),
+                    quantity=int(item.get('hldg_qty', 0) or 0),
+                    avgPrice=float(item.get('pchs_avg_pric', 0) or 0),
+                    currentPrice=float(item.get('prpr', 0) or 0),
+                    profit=float(item.get('evlu_pfls_amt', 0) or 0),
+                    profitRate=float(item.get('evlu_pfls_rt', 0) or 0),
+                    marketValue=float(item.get('evlu_amt', 0) or 0)
+                )
+                db.add(h)
+                
+            await db.commit()
+            logger.info(f"SyncService: Account Sync Complete. Assets: {total_assets}, Holdings: {len(output1)}")
+            
+        except Exception as e:
+            await db.rollback()
+            logger.error(f"SyncService: Error during sync: {e}")
+
+sync_service = SyncService()