해외선물 자동매매 시스템 활용법
709 조회
0
추천
0
비추천
1. 자동매매 시스템이란?
자동매매 시스템은 컴퓨터 프로그램이 사전에 설정한 전략에 따라 매수·매도를 자동으로 실행하는 방식입니다.
대표적으로 HTS API, MT4/MT5, NinjaTrader, Tradestation, 파이썬 기반 시스템 등이 활용됩니다.
※ 장점
＊ 감정 배제 → 일관된 매매 가능
＊ 24시간 시장 모니터링 가능
＊ 백테스트를 통한 전략 검증 가능
※ 단점
＊ 시스템 오류/인터넷 장애 리스크
＊ 전략 미검증 시 대규모 손실 가능
2. 준비 단계
(1) 거래 플랫폼 선택
＊ MT5 / MT4: 글로벌 브로커에서 널리 사용, EA(Expert Advisor) 기반 자동매매
＊ 파이썬 + API: 증권사 API(키움증권, 대신증권) 또는 Interactive Brokers API 사용
＊ NinjaTrader / MultiCharts: 전문 트레이더용, 강력한 차트·전략 백테스트 기능
(2) 전략 설계
＊ 지표 기반 전략: 이동평균선 골든크로스, RSI 과매수/과매도 등
＊ 가격 행동 기반: 특정 가격 돌파, 캔들 패턴 매매
＊ 복합 전략: 지표+가격 패턴 결합
(3) 데이터 확보
＊ 거래소 티커 데이터, 과거 시세(OHLCV), 경제 지표
＊ 데이터 정확성이 전략 성과에 직결됨
3. 자동매매 구현 방법
(1) 백테스트 (Backtest)
＊ 과거 데이터로 전략의 수익률·승률·DD(최대 손실폭) 확인
＊ 최소 2~3년 데이터로 검증, 과최적화(Overfitting) 방지
(2) 모의매매 (Paper Trading)
＊ 실시간 가상 거래 환경에서 전략 테스트
＊ 슬리피지, 수수료 반영 필수
(3) 실거래 적용
＊ 초기엔 소규모 계약으로 테스트
＊ 전략 성과 모니터링 후 점진적 확대
4. 리스크 관리 필수 요소
1) 손절·익절 자동 설정
＊ 코드에서 포지션 종료 조건 강제
2) 최대 손실 한도 설정
＊ 하루/주간 단위로 손실 한도 도달 시 자동 중단
3) 전략 다변화
＊ 한 종목·한 전략에 집중하지 않고 분산 운용
4) 서버 안정성 확보
＊ VPS(가상 서버) 사용으로 24시간 연결 유지
＊ 인터넷/전원 장애 대비
5. 활용 팁
＊ 전략 코드 오픈소스 활용: GitHub, TradingView Pine Script에서 참조
＊ 모듈화: 진입·청산 로직을 별도 함수로 관리 → 유지보수 용이
＊ 심리 관리: 자동매매라도 주기적 모니터링으로 비정상 상황 대응
※ 중요 포인트 ※
자동매매는 "버튼만 누르면 수익"이 아니라,
전략 → 검증 → 리스크 관리라는 3단계를 제대로 거쳐야 장기적으로 계좌를 지킬 수 있습니다.
해외선물 자동매매 3종 파이썬 예제 코드
＊ 예제는: MT5나 키움 API에 연결 가능한 형태 입니다 직접 설정이 가능합니다.
해외선물 자동매매 3종 파이썬 예제 코드
＊ 예제는: MT5나 키움 API에 연결 가능한 형태 입니다 직접 설정이 가능합니다.
코드는 MetaTrader5(mt5) 라이브러리 사용 예제로 작성했으며(가장 간단히 실계좌/모의계좌와 연동 가능), 공통 플랫폼 프레임워크 형태로 제공됩니다.
★ 포인트: 세 전략은 모두 같은 프레임워크(데이터 로드 → 시그널 생성 → 리스크 관리 → 주문 실행 → 로깅)를 공유하므로, 나중에 다른 전략 추가 · 다른 브로커 API로 포팅하기 쉽습니다.
☆ 주의: 실거래 전 반드시 백테스트(백테스터) → 모의 실시간(페이퍼) 테스트를 충분히 하세요. 코드 그대로 실계좌에 돌리면 손실이 날 수 있습니다.
파일 구조 (추천)
bash
auto_futures/
├─ config.py
├─ utils.py
├─ strategies.py
├─ executor.py # 메인: 연결, 루프, 전략 선택
└─ backtest.py # 간단한 백테스터(옵션)
아래는 각 파일의 핵심 코드(줄여서 제공). 복사해서 바로 실행 가능한 형태로 합쳐도 됩니다.
config.py — 환경설정 / 리스크 파라미터
python
# config.py
SYMBOL = "NQmini" # 거래 심볼 (MT5 심볼명에 맞게 수정)
TIMEFRAME = "M5" # M1, M5, M15 등 (데이터 수집 기준)
LOT_SIZE = 1 # 계약 수 (계약 단위는 심볼에 따라 다름)
ACCOUNT_RISK = 0.01 # 계좌에서 1회 거래 리스크 비율 (예: 1%)
MAX_DAILY_LOSS = 0.03 # 하루 최대 손실(계좌 대비)
SLIPPAGE = 2 # 슬리피지(틱 단위 또는 가격 단위, 브로커에 따라)
MAX_POSITION = 3 # 최대 동시 포지션 수
LOG_FILE = "trade_log.csv"
VPS = True # True면 VPS 가정(연결 안정)
utils.py — 공통 유틸 (포지션 사이징, 로깅 등)
python
# utils.py
import math, csv, time
from datetime import datetime
def calc_lot(account_balance, risk_pct, entry_price, stop_price, tick_value_per_contract=5):
# 간단 포지션 사이징 예시 (틱 기반)
# 손실 허용액
risk_amount = account_balance * risk_pct
# 가격 차이 (절대값)
price_diff = abs(entry_price - stop_price)
if price_diff == 0:
return 0
# 예시: tick_value_per_contract는 1틱당 가치(종목마다 다름)
# 계약 수 = risk_amount / (price_diff * tick_value_per_contract)
lots = risk_amount / (price_diff * tick_value_per_contract)
# 정수 계약 단위로 반올림
return max(1, int(math.floor(lots)))
def log_trade(row, filename="trade_log.csv"):
header = ["timestamp","symbol","strategy","action","price","volume","sl","tp","reason"]
write_header = False
try:
with open(filename, "r") as f: pass
except FileNotFoundError:
write_header = True
with open(filename,"a",newline='') as f:
writer = csv.writer(f)
if write_header:
writer.writerow(header)
writer.writerow(row)
def now_str():
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
strategies.py — 3개 전략의 시그널 함수
python
# strategies.py
import pandas as pd
import talib as ta
def ma_crossover(df, short=9, long=21):
"""
이동평균 크로스 전략 (단순)
df: DataFrame with 'close'
returns: signal: "BUY"/"SELL"/None and SL/TP (price)
"""
close = df['close'].values
ma_s = ta.SMA(close, timeperiod=short)
ma_l = ta.SMA(close, timeperiod=long)
if len(ma_s) < 2 or len(ma_l) < 2:
return None, None, None
# 직전과 현재 교차 체크
prev_cross = ma_s[-2] - ma_l[-2]
now_cross = ma_s[-1] - ma_l[-1]
# 골든크로스
if prev_cross <= 0 and now_cross > 0:
entry = close[-1]
sl = min(df['low'].rolling(5).min().iloc[-1], entry - (0.5 * (entry*0.005))) # 예시 SL
tp = entry + 2*(entry - sl)
return "BUY", sl, tp
# 데드크로스
if prev_cross >= 0 and now_cross < 0:
entry = close[-1]
sl = max(df['high'].rolling(5).max().iloc[-1], entry + (0.5 * (entry*0.005)))
tp = entry - 2*(sl - entry)
return "SELL", sl, tp
return None, None, None
def breakout(df, lookback=20, atr_mult=1.5):
"""
단순 변동성 브레이크아웃
진입: 과거 n봉의 고가 돌파 (long) / 저가 돌파 (short)
SL: 진입가 - ATR*atr_mult
TP: 위험:보상 비율 1:2
"""
high = df['high'].values
low = df['low'].values
close = df['close'].values
if len(df) < lookback+2:
return None, None, None
recent_high = max(df['high'].iloc[-(lookback+1):-1])
recent_low = min(df['low'].iloc[-(lookback+1):-1])
entry = close[-1]
atr = ta.ATR(df['high'], df['low'], df['close'], timeperiod=14).iloc[-1]
if entry > recent_high:
sl = entry - atr * atr_mult
tp = entry + (entry - sl)*2
return "BUY", sl, tp
if entry < recent_low:
sl = entry + atr * atr_mult
tp = entry - (sl - entry)*2
return "SELL", sl, tp
return None, None, None
def rsi_mean_reversion(df, rsi_period=14, overbought=70, oversold=30):
"""
RSI 평균회귀: 과매수/과매도 레벨에서 반대 포지션
진입 후 SL은 최근 고저 기반
"""
close = df['close']
rsi = ta.RSI(close, timeperiod=rsi_period)
if len(rsi) < 2:
return None, None, None
cur_rsi = rsi.iloc[-1]
entry = close.iloc[-1]
if cur_rsi >= overbought:
# 과매수 -> 숏
sl = df['high'].rolling(5).max().iloc[-1] + 0.5*(entry*0.001)
tp = entry - (sl - entry)*1.5
return "SELL", sl, tp
if cur_rsi <= oversold:
sl = df['low'].rolling(5).min().iloc[-1] - 0.5*(entry*0.001)
tp = entry + (entry - sl)*1.5
return "BUY", sl, tp
return None, None, None
executor.py — MT5 연결, 메인 루프, 주문 실행 (핵심)
python
# executor.py
import MetaTrader5 as mt5
import pandas as pd, time
from config import SYMBOL, TIMEFRAME, LOT_SIZE, ACCOUNT_RISK, MAX_DAILY_LOSS, SLIPPAGE, LOG_FILE
from utils import calc_lot, log_trade, now_str
from strategies import ma_crossover, breakout, rsi_mean_reversion
# --- MT5 연결 함수 예시 ---
def connect_mt5(login=None, password=None, server=None):
if not mt5.initialize():
print("MT5 initialize() failed")
return False
# 만약 계정 로그인 필요 시
if login:
authorized = mt5.login(login, password=password, server=server)
if not authorized:
print("MT5 login failed")
return False
return True
# --- 데이터 로드 (심볼/타임프레임에 맞춰) ---
def get_ohlcv(symbol, timeframe=mt5.TIMEFRAME_M5, n=500):
rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, n)
if rates is None:
return None
df = pd.DataFrame(rates)
df['time'] = pd.to_datetime(df['time'], unit='s')
df.rename(columns={'close':'close','open':'open','high':'high','low':'low','tick_volume':'volume'}, inplace=True)
return df
# --- 주문 함수 (단순 예시) ---
def place_order(symbol, action, volume, price, sl, tp):
# action: "BUY" or "SELL"
# 이 함수는 단순 예시로 실제 주문 파라미터는 브로커별로 다름
order_type = mt5.ORDER_TYPE_BUY if action=="BUY" else mt5.ORDER_TYPE_SELL
request = {
"action": mt5.TRADE_ACTION_DEAL,
"symbol": symbol,
"volume": float(volume),
"type": order_type,
"price": price,
"sl": sl,
"tp": tp,
"deviation": SLIPPAGE,
"magic": 123456,
"comment": "auto_strategy"
}
result = mt5.order_send(request)
return result
# --- 메인 루프 (단순화) ---
def run(strategy_name="ma"):
# connect
if not connect_mt5():
return
# map strategy
strat_map = {"ma": ma_crossover, "breakout": breakout, "rsi": rsi_mean_reversion}
strat_fn = strat_map.get(strategy_name)
if not strat_fn:
print("Unknown strategy")
return
# account info
account_info = mt5.account_info()
balance = account_info.balance
print("Account balance:", balance)
# main polling loop (예: 분봉 완료 시마다 실행)
try:
while True:
df = get_ohlcv(SYMBOL, mt5.TIMEFRAME_M5, n=500)
if df is None or df.empty:
time.sleep(5); continue
signal, sl, tp = strat_fn(df)
if signal:
entry_price = df['close'].iloc[-1]
lot = calc_lot(balance, ACCOUNT_RISK, entry_price, sl, tick_value_per_contract=5)
# safety checks
if lot <= 0:
print("Calculated lot 0 -> skip")
else:
res = place_order(SYMBOL, signal, lot, entry_price, sl, tp)
print(now_str(), "Order result:", res)
log_trade([now_str(), SYMBOL, strategy_name, signal, entry_price, lot, sl, tp, "signal"])
# 주기: 타임프레임과 맞춰 조정
time.sleep(60) # 간단 예시
except KeyboardInterrupt:
print("Stopped by user")
finally:
mt5.shutdown()
if __name__ == "__main__":
# 예시: MA 전략 실행
run(strategy_name="ma")
backtest.py — 간단한 백테스터 (전략 성능 빠르게 체크)
python
# backtest.py (간단 샘플)
import pandas as pd
from strategies import ma_crossover
from utils import calc_lot
def simple_backtest(df, strat_fn, starting_balance=1000000):
balance = starting_balance
trades = []
for i in range(50, len(df)):
window = df.iloc[:i+1].copy()
signal, sl, tp = strat_fn(window)
if signal:
entry = window['close'].iloc[-1]
lot = calc_lot(balance, 0.01, entry, sl, tick_value_per_contract=5)
if lot <= 0:
continue
# 단순 가정: 다음 봉에서 tp 또는 sl 도달 여부 체크 (단순 시뮬)
nxt_high = df['high'].iloc[i+1] if i+1 < len(df) else entry
nxt_low = df['low'].iloc[i+1] if i+1 < len(df) else entry
pnl = 0
if signal=="BUY":
if nxt_high >= tp:
pnl = (tp - entry) * lot * 5
elif nxt_low <= sl:
pnl = (sl - entry) * lot * 5
else:
pnl = (df['close'].iloc[i+1] - entry) * lot * 5
else:
if nxt_low <= tp:
pnl = (entry - tp) * lot * 5
elif nxt_high >= sl:
pnl = (entry - sl) * lot * 5
else:
pnl = (entry - df['close'].iloc[i+1]) * lot * 5
balance += pnl
trades.append(pnl)
return {"final_balance": balance, "trades": trades, "count": len(trades)}
설명 & 사용 팁 (핵심)
1). MT5 심볼명: 브로커마다 심볼명이 다릅니다. mt5.symbols_get()로 사용 가능한 심볼 확인하세요.
2). 틱 가치(tick_value_per_contract): 예시는 5달러/틱(나스닥 미니)로 설정했지만 실제 심볼별 가치가 반드시 달라집니다. 이 값이 포지션 사이징 결과에 직접 영향을 줍니다.
3). 포지션 사이징: 예제는 매우 단순화된 계산입니다. 실제로는 마진/레버리지/증거금 요건을 반영해야 합니다.
4). 슬리피지 & 수수료 반영: 백테스트·모의매매 시 반드시 수수료와 슬리피지를 반영하세요.
5). VPS & 모니터링: 실시간 자동매매는 VPS에서 24/7 운용 권장. 장애시 자동 이메일/텔레그램 알림 추가하세요.
6). 안전 스위치: 하루 손실 한도 초과 시 자동 정지, 서버 재시작시 자동 재접속, 최대 동시 포지션 제한 등 꼭 구현하세요.
7). 로깅 & 리플레이: 모든 주문·에러·잔고변동을 로그로 남기고 주기적으로 복기하세요.
-
등록일 2025.10.30
-
등록일 2025.09.21
-
등록일 2025.09.21
-
등록일 2025.09.21
댓글 0
등록된 댓글이 없습니다.
