Lập Trình Bot Giao Dịch Chứng Khoán Python Với API TCBS và DNSE

Trong bài viết này, chúng ta đi vào thực hành: kết nối API chứng khoán Việt Nam (TCBS, DNSE), lấy dữ liệu giá real-time và gửi lệnh giao dịch tự động bằng Python.

Chuẩn Bị Môi Trường

pip install requests pandas numpy python-dotenv

Tạo file .env để lưu credentials an toàn — không bao giờ hard-code token trong code:

TCBS_TOKEN=your_token_here
DNSE_USERNAME=your_email
DNSE_PASSWORD=your_password

1. Lấy Dữ Liệu Giá Với TCBS API

Dữ Liệu Lịch Sử OHLCV

import requests
import pandas as pd
from datetime import datetime, timedelta

def get_historical_data(symbol: str, days: int = 365) -> pd.DataFrame:
    to_date = int(datetime.now().timestamp())
    from_date = int((datetime.now() - timedelta(days=days)).timestamp())

    url = (
        f"https://apipubaws.tcbs.com.vn/stock-insight/v1/stock/bars-long-term"
        f"?ticker={symbol}&type=stock&resolution=D"
        f"&from={from_date}&to={to_date}"
    )
    response = requests.get(url, timeout=10)
    data = response.json()

    df = pd.DataFrame(data['data'])
    df['date'] = pd.to_datetime(df['tradingDate'])
    return df.set_index('date').sort_index()

df_vnm = get_historical_data('VNM', days=365)
print(df_vnm.tail())

Giá Real-time

def get_realtime_price(symbol: str) -> dict:
    url = f"https://apipubaws.tcbs.com.vn/stock-insight/v1/stock/quotes?symbol={symbol}"
    r = requests.get(url).json()
    return {
        'symbol': symbol,
        'price': r['data'][0]['closePrice'],
        'volume': r['data'][0]['totalVolume'],
        'change_pct': r['data'][0]['priceChange']
    }

2. Đặt Lệnh Với DNSE API

DNSE cung cấp API đặt lệnh chính thức. Sau khi đăng ký và được cấp API access:

class DNSETrader:
    BASE_URL = "https://api.dnse.com.vn/v2"

    def __init__(self, access_token: str):
        self.headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json"
        }

    def place_order(self, symbol: str, side: str, quantity: int,
                    price: float, order_type: str = "LO") -> dict:
        payload = {
            "symbol": symbol,
            "side": side.upper(),
            "quantity": quantity,
            "price": price,
            "orderType": order_type
        }
        r = requests.post(f"{self.BASE_URL}/orders",
                          json=payload, headers=self.headers)
        return r.json()

3. Bot MA Crossover Tự Động Hoàn Chỉnh

import time

def is_trading_hours() -> bool:
    now = datetime.now()
    t = now.hour * 60 + now.minute
    morning = 9*60 <= t <= 11*60+30
    afternoon = 13*60 <= t <= 14*60+45
    return (morning or afternoon) and now.weekday() < 5

def run_bot(symbol: str, trader: DNSETrader):
    position = 0

    while True:
        if not is_trading_hours():
            time.sleep(60)
            continue

        df = get_historical_data(symbol, days=60)
        df['ema9'] = df['close'].ewm(span=9).mean()
        df['ema21'] = df['close'].ewm(span=21).mean()

        cross_up = (df['ema9'].iloc[-1] > df['ema21'].iloc[-1] and
                    df['ema9'].iloc[-2] < df['ema21'].iloc[-2])
        cross_down = (df['ema9'].iloc[-1] < df['ema21'].iloc[-1] and
                      df['ema9'].iloc[-2] > df['ema21'].iloc[-2])

        price = get_realtime_price(symbol)['price']

        if cross_up and position == 0:
            trader.place_order(symbol, 'buy', 100, price)
            position = 1
            print(f"BUY {symbol} @ {price}")
        elif cross_down and position == 1:
            trader.place_order(symbol, 'sell', 100, price)
            position = 0
            print(f"SELL {symbol} @ {price}")

        time.sleep(60)