코인선물 ai 자동매매 프로그램만들기 3강. 소스 코드 바이비트 api 로 주문 넣기

 

 

 

아래 코드는 ai 로 분석받은 결과로 바이비트 perpetual short, long 주문을 실행하는 기본 코드입니다. 아래 코드만으로도 ai 자동매매가 가능합니다.

 

발급받은 api 키를 입력하고, 거리, 수익, 배율을 결정해서 입력해야 작동합니다.

 

 

여기서 잠깐!

바이비트 22566 레퍼럴 가입 후 이 글에 비공개 댓글로 UID를 남겨주시면 

 

자동으로 스탑로스 가격이 지정되는 기능

TP 설정이 있다면 취소하고 거래 수수료 절약을 위해  limit 으로 재주문 입력되는 기능

long short open 주문 후 5분이 지나도 주문이 체결되지 않은 주문은 자동 취소되는 기능 (5분 반복 작업시 5분 캔들 2개 지난후 취소)

주문입력 후 주문내용이 csv 에 저장되는 기능이 있는 코드 +@ 

9개월간 투자 결과 알아낸 최적의 거리와 수익 비율을 공유 해드리겠습니다.  감사합니다.

 

import pybit
from pybit.unified_trading import *
import datetime
import time
import pandas as pd
import os

# API 키 설정
session = HTTP(
    testnet=False,
    api_key="발급받은 api 키를 여기에 입력하세요",
    api_secret="secret 키를 입력하세요",
    recv_window=60000  # recv_window 설정
)

original_time = time.time

# 서버 시간 가져오기
def get_server_time():
    response = session.get_server_time()
    if response['retCode'] == 0:
        return int(response['time'])
    else:
        return None

def get_time_difference():
    server_time_ms = get_server_time()
    if server_time_ms:
        local_time_ms = int(original_time() * 1000)
        return (server_time_ms - local_time_ms) / 1000.0
    return 0

time_diff = get_time_difference()

def adjusted_time():
    return original_time() + time_diff

time.time = adjusted_time

심볼 = "DOGEUSDT"
수량 = 0
가격소숫점자리 = 5
거리 = # 현재가와 얼마나 차이나는 가격에 주문가를 넣을지 설정
수익 = # 주문가와 얼마나 차이나는 가격에 수익을 낼지 설정

def get_latest_price(symbol):
    try:
        response = session.get_tickers(category="linear", symbol=symbol)
        if response['retCode'] == 0:
            return float(response['result']['list'][0]['lastPrice'])
        else:
            return None
    except:
        return None

def calculate_qty(totalEquity, symbol):
    latest_price = get_latest_price(symbol)
    if latest_price is not None:
        return round((totalEquity / latest_price) * 3 / 15, 0)
    raise ValueError("가격 정보 없음")

def get_current_positions(symbol):
    try:
        response = session.get_positions(category="linear", symbol=symbol)
        if response['retCode'] == 0:
            return response['result']['list']
        else:
            return []
    except:
        return []

def send_order(order_type, price, qty, entry_price=None):
    최소주문금액 = 5
    주문가격 = price * qty
    if 주문가격 < 최소주문금액:
        qty = round(최소주문금액 / price, 0)
    try:
        side = "Buy" if order_type == 'long' else "Sell"
        price = round(price * (1 - 거리), 가격소숫점자리) if order_type == 'long' else round(price * (1 + 거리), 가격소숫점자리)
        tp_price = round(price * (1 + 수익) if order_type == 'long' else price * (1 - 수익), 가격소숫점자리)
        response = session.place_order(
            category="linear",
            symbol=심볼,
            side=side,
            orderType="Limit",
            qty=round(qty, 1),
            price=price,
            timeInForce="PostOnly",
            reduceOnly=False,
            closeOnTrigger=False,
            positionIdx=1 if order_type == 'long' else 2,
            takeProfit=tp_price,
            tpslMode="Partial",
            recv_window=60000
        )
        return response
    except Exception as e:
        print(f"{order_type} 주문 예외 발생: {e}")
        return None

def execute_trade(order_type):
    try:
        latest_price = get_latest_price(심볼)
        if latest_price:
            entry_price = None
            current_positions = get_current_positions(심볼)
            if current_positions:
                for position in current_positions:
                    if position['side'].lower() == 'buy' and order_type == 'long':
                        entry_price = float(position.get('avgPrice', 0))
                    elif position['side'].lower() == 'sell' and order_type == 'short':
                        entry_price = float(position.get('avgPrice', 0))
            send_order_response = send_order(order_type, latest_price, 기본_qty, entry_price)
    except Exception as e:
        print(f"거래 실행 예외 발생: {e}")

def get_last_position_from_csv():
    try:
        current_directory = os.path.dirname(os.path.abspath(__file__))
        csv_path = os.path.join(current_directory, '도지market_analysis_results.csv')
        df = pd.read_csv(csv_path)
        last_position = df.iloc[-1]['position']
        return last_position.strip().lower()
    except Exception as e:
        print(f"CSV 파일 읽기 예외 발생: {e}")
        return None

if __name__ == "__main__":
    try:
        wallet_info = session.get_wallet_balance(
            accountType="UNIFIED",
            coin="USDT",
            recv_window=60000
        )
        totalEquity = float(wallet_info['result']['list'][0]['totalEquity'])
        기본_qty = calculate_qty(totalEquity, 심볼)

        last_position = get_last_position_from_csv()

        if last_position == 'long':
            execute_trade('long')
        elif last_position == 'short':
            execute_trade('short')
        else:
            print(f"유효하지 않은 마지막 포지션: {last_position}")
    except Exception as e:
        print(f"메인 로직 예외 발생: {e}")