首页 > 其他分享 >海龟交易系统的实盘部署

海龟交易系统的实盘部署

时间:2024-03-03 23:22:47浏览次数:25  
标签:实盘 交易系统 self float ks position 海龟 close longLevel

近期(20240301)加密货币市场涨幅喜人,不断有新资金流入。笔者之前过对趋势交易有过一段时间研究,各种策略都略有了解,但在仓位管理和风险控制上都不太精通,比较幸运的是有一公开的趋势跟踪策略不仅可以跟踪趋势 还有基于波动率的风险控制方法,被称为海龟交易系统,在上个世纪曾经风靡一时。曾经用python写过一个海龟交易(原封不动的复刻的经典策略),但是没有对接实盘,今天就利用okx平台的接口接入实盘吧。

注意 这个海龟策略,有相当长的历史了 ,很多内容已经过时了 ,用于短期套利 完全没问题,毕竟是正宗的趋势交易法 盈亏比极大, 但是要长期投资的话 必然大亏! 投资上记住,收益的90%都来自正确的资产配置,这个道理已经被经济学家验证很多次了。

点击查看代码
import asyncio
import datetime
import json
import time

from okx import MarketData, Trade, Account
from okx.websocket.WsPublicAsync import WsPublicAsync

id = "PEPE-USDT"
bar = "1m"


class Kline:
    def __init__(self, max_line=55, before='', after=''):
        # API 初始化
        apikey = "*******"
        secretkey = "******"
        passphrase = "****"

        flag = "1"  # 实盘: 0, 模拟盘: 1
        self.accountAPI = Account.AccountAPI(apikey, secretkey, passphrase, False, flag)
        self.tradeAPI = Trade.TradeAPI(apikey, secretkey, passphrase, False, flag)

        self.n = None
        self.max_units = 6
        self.max_line = max_line

        md = MarketData.MarketAPI(flag='1', debug=False)
        ret = md.get_mark_price_candlesticks_chai(instId=id, limit=max_line, bar=bar, before=before, after=after)

        self.ks = ret['data'][0:max_line]
        print(f"初始化k线:{self.ks}")
        #  由rest请求构建 由websocket维护
        # atr(20) 10k高低线 20k高低线 55k高低线 平均持仓成本 前一订单的持仓成本 止损是atr的n倍数 浮盈前订单+0.5atr加仓
        self.atr = None  # atr
        self.pyramid_input = 1  # 加仓的atr倍数
        self.stop_input = 4  # 止损的atr倍数

        self.l10 = self.h10 = self.l20 = self.h20 = self.l55 = self.h55 = 0
        self.high = self.low = self.close = self.open = 0.0

        self.position_avg_price = None  # 市场的平均入场价
        self.netprofit = 0.0  # 已完成交易的利润
        self.position_size = 0.0  # 头寸的大小
        self.position_price = 0.0  # 头寸的总成本 (持仓的市值)

        # 查看账户余额
        result2 = self.accountAPI.get_account_balance()
        money=0.0
        for i  in result2["data"][0]["details"]:
            if i["ccy"]=="USDT":
                money=float(i["cashBal"])
                print(money)
        if money==0.0:
            exit('money error')
        self.capitalLeft = money  # 总资金
        self.initial_capital = money  # 策略初始资金

        self.riskPercent = 0.002  # 单笔交易承受的风险 越高持仓越多 加仓越少

        self.exit_long = None
        self.enter_long = None

        self.risk = 0.01
        self.win = False  # 上次交易是否获利

        self.buyPrice = 0.0  # 上次做多的价位

        self.nextBuyPrice = 0.0  # 下一次做多的价位

        self.stopPrice = 0.0  # 止损价

        self.totalBuys = 0  # 金字塔加仓的总数

        self.inBuy = False  # 是否在一个做多的头寸
        self.inBuy_p = False  # 是否在一个做多的头寸

        self.longLevel = None

        self.mode = 'L1'
        self.mode_p = 'L1'

        self.fake = False
        self.fake_p = False

        self.fakeBuyPrice = 0.0  # 假交易的价位

        self.shares = 0.0

        self.sxf = 0.0  # 手续费

    # o h l c
    # ['1709080200000', '57031.1', '57107.4', '57031', '57102.7', '63367', '63.367', '3616608.9098', '0']
    def buy(self, size):
        msg = self.tradeAPI.place_order(instId=id,
                                        tdMode="cash",
                                        clOrdId="b15",
                                        side="buy",
                                        ordType="market",
                                        sz=f"{size}")
        return msg

    def sell(self):
        # 市价全平
        result = self.accountAPI.get_max_order_size(
            instId=id,
            tdMode="isolated"
        )

        sell = float(result["data"][0]['maxSell'])

        msg = self.tradeAPI.place_order(instId=id,
                                        tdMode="cash",
                                        clOrdId="b15",
                                        side="sell",
                                        ordType="market",
                                        sz=f"{sell}")
        print(msg)
        self.netprofit = 0.0  # 已完成交易的利润
        self.position_size = 0.0  # 头寸的大小
        self.position_price = 0.0  # 头寸的总成本 (持仓的市值)
        # 查看账户余额
        result2 = self.accountAPI.get_account_balance()

        money=self.initial_capital
        for i  in result2["data"][0]["details"]:
            if i["ccy"]=="USDT":
                money=float(i["cashBal"])
                print(money)
        self.capitalLeft = money  # 总资金
        self.initial_capital = money  # 策略初始资金

    def close(self):

        pass

    def run_s(self):

        self.high, self.low, self.close, self.open = float(self.ks[0][2]), float(self.ks[0][3]), float(
            self.ks[0][4]), float(self.ks[0][1])

        self.inBuy_p = self.inBuy
        self.mode_p = self.mode
        self.fake_p = self.fake

        # 判断是否进入做多头寸
        if not self.inBuy and (self.high > self.h20 or self.high > self.h55):
            self.inBuy = True
        else:
            # 判断是否需要退出做多头寸
            if self.inBuy:
                if self.mode == 'L1' and self.low < self.l10:
                    self.inBuy = False
                elif self.mode == 'L2' and self.low < self.l20:
                    self.inBuy = False
                elif self.low < self.stopPrice:
                    self.inBuy = False

        # 如果没有头寸 且突破 上次盈利 就标记为虚假头寸
        if not self.inBuy_p and self.high > self.h20 and self.win:
            self.fake = True
            self.fakeBuyPrice = self.close

        # 当上个为假订单,在此时平仓,关闭假订单 并根据盈利情况赋值给win
        if self.fake_p and self.inBuy_p and not self.inBuy:
            self.fake = False
            self.win = self.close >= self.fakeBuyPrice

        # 突破h55时始终为真订单
        self.fake = False if self.high > self.h55 else self.fake

        # 当突破l1 2时记录多头水平
        self.longLevel = 'L1' if not self.inBuy_p and self.high > self.h20 else None
        if (not self.inBuy_p or (self.inBuy_p and self.fake)) and self.high > self.h55:
            self.longLevel = 'L2'

        if self.longLevel is not None:
            self.mode = self.longLevel

        if self.longLevel in ['L1', 'L2']:
            self.buyPrice = self.close
            self.totalBuys = 1
            self.stopPrice = self.close - (self.stop_input * self.n)
            self.nextBuyPrice = self.close + (self.pyramid_input * self.n)

        # 当加仓时
        if self.longLevel is None and self.inBuy_p and self.high > self.nextBuyPrice and self.totalBuys < self.max_units:
            self.longLevel = 'P'
            self.buyPrice = self.close
            self.totalBuys += 1
            self.stopPrice = self.close - (self.stop_input * self.n)
            self.nextBuyPrice = self.close + (self.pyramid_input * self.n)

        # Tracks stops and exits, marking them with SG or SR
        if self.position_avg_price is not None:
            if self.longLevel is None and self.inBuy_p and self.low < self.stopPrice and self.close >= self.position_avg_price:
                self.longLevel = 'SG'
            elif self.longLevel is None and self.inBuy_p and self.low < self.stopPrice and self.close < self.position_avg_price:
                self.longLevel = 'SR'
            elif self.longLevel is None and self.mode_p == 'L1' and self.inBuy_p and (
                    self.low < self.l10) and self.close >= self.position_avg_price:
                self.longLevel = 'SG'
            elif self.longLevel is None and self.mode_p == 'L2' and self.inBuy_p and (
                    self.low < self.l20) and self.close >= self.position_avg_price:
                self.longLevel = 'SG'
            elif self.longLevel is None and self.mode_p == 'L1' and self.inBuy_p and (
                    self.low < self.l10) and self.close < self.position_avg_price:
                self.longLevel = 'SR'
            elif self.longLevel is None and self.mode_p == 'L2' and self.inBuy_p and (
                    self.low < self.l20) and self.close < self.position_avg_price:
                self.longLevel = 'SR'

        # Tracks if the trade was a win or loss.
        if self.longLevel == 'SG':
            self.win = True
        if self.longLevel == 'SR':
            self.win = False

        # Variables used to tell strategy when to enter/exit trade.
        self.enter_long = (self.longLevel == 'L1' or self.longLevel == 'L2' or self.longLevel == 'P') and not self.fake
        self.exit_long = (self.longLevel == 'SG' or self.longLevel == 'SR') and (not self.fake)

        self.risk = (self.initial_capital + self.netprofit) * self.riskPercent
        self.shares = float(self.risk / (self.stop_input * self.n))

        if self.position_avg_price is not None:
            self.capitalLeft = self.initial_capital + self.netprofit - (self.position_size * self.position_avg_price)
        else:
            self.capitalLeft = self.initial_capital + self.netprofit

        if self.shares * self.close > self.capitalLeft:
            self.shares = max(0.0, float(self.capitalLeft / self.close))

        self.shares = max(0.0, self.shares)

        # 做多时更新入场价
        if self.enter_long:
            self.position_size += self.shares  # 头寸的大小
            self.position_price += self.close * self.shares  # 头寸成本增加
            self.position_avg_price = self.position_price / self.position_size  # 市场的平均入场价
            print(
                f"多{self.shares}张,price:{self.close},占:{self.shares * self.close / self.initial_capital * 100}%,{datetime.datetime.fromtimestamp(int(self.ks[0][0]) / 1000.0)}")
            self.sxf += (self.shares * self.close) * 0.0008
            if self.shares != 0.0:
                self.buy(int(self.shares * self.close))
                pass
        if self.exit_long:
            print(
                f"平{self.position_size}张,price:{self.close},{datetime.datetime.fromtimestamp(int(self.ks[0][0]) / 1000.0)}")

            self.netprofit += (self.close - self.position_avg_price) * self.position_size  # 已完成交易的利润
            self.position_size = 0.0  # 头寸的大小
            self.position_avg_price = None  # 市场的平均入场价

            self.position_price = 0.0  # 头寸sum
            print(f"交易盈利:{self.netprofit},手续费{self.sxf}")
            self.sell()

    # 指标数值更新
    def reload(self, k):
        if self.ks[0][0] == k[0][0]:
            self.ks[0] = k[0]
        else:
            self.ks.insert(0, k[0])

            del self.ks[self.max_line - 1]
            # 更新atr值
            self.atr = self.reload_atr(self.ks)
            self.n = self.atr

            # 更新各周期最大最小值
            self.l10, self.h10 = kline.hl_find(self.ks, 10)
            self.l20, self.h20 = kline.hl_find(self.ks, 20)
            self.l55, self.h55 = kline.hl_find(self.ks, 55)

        self.run_s()

    @staticmethod
    def hl_find(data, window):
        d = data[0:window]
        # o h l c
        # ['1709080200000', '57031.1', '57107.4', '57031', '57102.7', '63367', '63.367', '3616608.9098', '0']

        # 初始化最大值和最小值为第一个子列表的第2个元素 本次k线不算
        min_value = float(data[1][3])
        max_value = float(data[1][2])

        # 遍历剩余子列表,更新最大值和最小值
        for sublist in d[1:]:

            if float(sublist[3]) < min_value:
                min_value = float(sublist[3])
            if float(sublist[2]) > max_value:
                max_value = float(sublist[2])

        return min_value, max_value

    @staticmethod
    def calculate_sma(data, window):

        return sum(data[:window]) / window

    @staticmethod
    def reload_atr(ks):
        # TR = max(high - low, abs(high - close[1]), abs(low - close[1]))
        count = len(ks)
        tr = [0.0] * count
        # o h l c
        # ['1709080200000', '57031.1', '57107.4', '57031', '57102.7', '63367', '63.367', '3616608.9098', '0']
        for i in range(count - 1, -1, -1):
            if i == count - 1:
                tr[i] = float(ks[i][2]) - float(ks[i][3])
            else:
                tr[i] = max(float(ks[i][2]) - float(ks[i][3]), abs(float(ks[i][2]) - float(ks[i + 1][4])),
                            abs(float(ks[i][3]) - float(ks[i + 1][4])))
        # ATR = SMA(TR, n)

        # 计算SMA,时间段为3
        atr = kline.calculate_sma(tr, 20)

        return atr


def publicCallback(message):
    # 解析字符串为Python对象
    global kline
    message = json.loads(message)
    if 'data' in message:
        k = message['data']
        kline.reload(k)

    pass


async def main():
    # url = "wss://wspap.okex.com:8443/ws/v5/public?brokerId=9999"
    url = "wss://wspap.okx.com:8443/ws/v5/business?brokerId=9999"
    ws = WsPublicAsync(url=url)
    await ws.start()
    args = []

    arg1 = {"channel": "candle1m", "instId": f"{id}"}

    args.append(arg1)

    await ws.subscribe(args, publicCallback)
    await asyncio.sleep(9999999999999)

    print("-----------------------------------------unsubscribe all--------------------------------------------")
    await ws.unsubscribe(args, None)


if __name__ == '__main__':
    kline = Kline(55)
    asyncio.run(main())
    """
        md2 = MarketData.MarketAPI(flag='1', debug=False)
    before_s = 1708120700000
    # 从2023开始, before="1708099200000", after="1708120800000"
    kline = Kline(55, before="1707926400000", after="1708120800000")
    for i in range(100000):
        time.sleep(0.2)
        ret3 = \
            md2.get_mark_price_candlesticks_chai(instId=id, bar=bar, before=str(before_s),
                                                 after=str(before_s + 3600000))[
                'data']
        before_s += 3600000
        kline.reload(ret3)
    """





策略的逻辑还是有点复杂的,之前的代码是在模拟盘运行的 我运行了几个晚上 大概没什么问题,如果要实盘主要要注意一下程序代码和实盘账户的对接不要出错,下单前要再次检查账户资金是否够用,平仓后也要重新调整类中保存的账户信息,冗余的代码是有必要的 ,谁知道会不会因为看似不起眼的数据同步误差导致系统发错错误的买卖指令呢。 我加入以下代码 在各个环节再次检验
点击查看代码
    def buy(self, size):
        msg = self.tradeAPI.place_order(instId=id,
                                        tdMode="cash",
                                        clOrdId="b15",
                                        side="buy",
                                        ordType="market",
                                        sz=f"{size}")
        
        money = self.capitalLeft
        # 查看账户余额
        result2 = self.accountAPI.get_account_balance()
        for i  in result2["data"][0]["details"]:
            if i["ccy"]=="USDT":
                money=float(i["cashBal"])
                print(money)
        self.capitalLeft = money  # 总资金
        
        return msg

买完之后更新下持仓

还有 如果程序发错了错误的买卖指令或者明显不合理的指令 要有函数能拦截 不让指令发出 这个功能我是这样实现的

点击查看代码
    # size 是真实的买单金额
    def check(self,size,warning_value=0.5):
        if size>self.capitalLeft*warning_value:
            self.sell()
            exit("order size>warning_value !")

在买入指令下达前直接检测金额是否超过警戒值 如果超过 全部头寸卖出后退出程序

其实对接还是比较简单的 几乎没有难对 就是对着api文档编程 难点全在买卖系统的建模 传统海龟交易模型 有很大的漏洞 当交易成本(手续费,流动性)太大,会在大量的止损单中损失大部分本金,这些买卖模型的建模需要非常复杂的逻辑实现 现代的交易系统大多使用强化学习,统计分析等手段实现,这里不做过多解释,最后给出程序的流程图希望对大家的投机或者赌博有所帮助。


这个流图没画完 原版还有虚假头寸和自保险下单的逻辑还有仓位控制 逻辑太复杂 懒得画了,感兴趣参考最开始我写的代码吧

标签:实盘,交易系统,self,float,ks,position,海龟,close,longLevel
From: https://www.cnblogs.com/muci6/p/18050977

相关文章

  • python海龟时钟
    运行效果代码importturtleastimporttimeastmdefDrawDial():#绘制表盘ts=t.Turtle()ts.hideturtle()ts.pensize(20)ts.speed(0)ts.color("cyan","yellow")ts.penup()ts.goto(0,-200)ts.pendown()ts.b......
  • 通达信【红影】今买明卖T+1超短线神器 实盘图例 主图幅图选股公式 源码文件分享
    通达信红影主图幅图选股公式今买明卖T+1超短线神器实盘图例源码文件分享某地金钻指标,原价2000哦,原公式完全加密,股海网和谐源码文件分享所谓红影,就是出现长上影线后收的红色K线,长上影线,代表有很强的资金能量拉升,收红色K线代表多大占据优势,次日大概率有冲高动能,原理非常简单,很......
  • 大厂交易系统从0到1(01)-一笔交易到底涉及多少张表?
    退款逻辑,1个业务单号--关联多个支付单号--也关联多个退款单号?退款单号,关联优惠券返还的,也可在退款单列表关联吗?返还比例多少,是在优惠券系统设置?整个交易、支付、清结算、账务体系杂糅,会产生很多单据、单号。再考虑正向、逆向,他们的关系更复杂。本文就来搞定订单、账单、支付记录......
  • 如何构建适合自己的交易系统
    交易者若想在股票投资中长期稳定地获胜,必须成功地解决两大难题:①如何在高度随机的价格波动中寻找到非随机性的部分。②如何有效地控制自身的心理弱点,使之不致影响自己的理性决策。想要做到这两点,你需要建立一套稳定的交易体系来让交易有章法可依,一个优秀的交易系统可以帮助交易者......
  • 合约交易系统开发(开发案例)丨合约交易所系统开发(逻辑方案)/设计程序/成熟源码
    智能合约的生命周期根据其运行机制可概括为协商、开发、部署、运维、学习和自毁六个阶段,其中开发阶段包括合约上链前的合约测试,学习阶段包括智能合约的运行反馈与合约更新为智能合约的基础架构模型,模型自底向上由基础设施层、合约层、运维层、智能层、表现层和应用层组成,基......
  • DAPP智能合约交易系统定制开发
     DAPP智能合约交易系统开发、dapp合约交易所开发,dapp合约交易系统软件开发智能合约系统根据事件描述信息中包含的触发条件,当触发条件满足时,从智能合约自动发出预设的数据资源,以及包括触发条件的事件;整个智能合约系统的核心就在于智能合约以事务和事件的方式经过智能合约模块的处理......
  • 中科驭数+光大证券,全国产化极速交易系统,生产环境上线230天!
    基于中科驭数KPUSWIFT®-2200NPro低时延网络DPU卡构建的极速交易系统,已在光大证券生产环境稳定上线运行超230天!基于双方的合作,光大证券完成了行业内首次基于国产CPU+国产低时延网卡+顶点软件的信创环境,实现在生产环境中从OMS到交易所的全业务流程打通,对TDGW实现即插即用,成功攻克......
  • 基于java斗车交易系统设计与实现
    21世纪的今天,随着社会的不断发展与进步,人们对于信息科学化的认识,已由低层次向高层次发展,由原来的感性认识向理性认识提高,管理工作的重要性已逐渐被人们所认识,科学化的管理,使信息存储达到准确、快速、完善,并能提高工作管理效率,促进其发展。论文主要是对斗车交易系统进行了介绍,包括......
  • 分享一套 MT4 crm MT4 MT5 CRM源码、web trade交易系统
    一套MT4MT5CRM源码,有跟单社区,同时支持MT4进行对接使用,支持代理返佣自由进行设置,可自动实时同步manager后台分组、交易品种和客户所有信息。包括带有内部实时内转功能,支持任何第三方支付、区块链和电子钱包。整套系统功能齐全。可节约公司大量租用成本和防止第三方公司泄露客户资......
  • 合约量化交易系统APP开发案例
    项目背景:该案例软件的开发是一款针对市场交易员而设计的软件系统,它能帮助交易者实现高效率的合约交易。我们目的是以用户的需求为主进行开发,应对市场上未知的风险和把控。开发过程:软件在开发过程中,采用了多种的开发按技术,软件工具包,开发框架,数据库技术,移动端的开发。在开发前先......