技術標籤：量化交易python

策略思路：
1、構建這樣一個函式，縱軸代表波幅，橫軸代表時間。從60分鐘前開始到現在共1小時，若60分鐘內漲跌幅>10，50分鐘內漲跌幅>9.96，40分鐘內漲跌幅>7.54，依此類推，1分鐘內漲跌幅超過約2.5美元，則發出異常波動提示，同時計算多空指數。
指數計算方法為z=(3-(y-20)²/200)(1.5-(time-1)/50)，這裡的time=60-x，例如一分鐘內漲幅為3美元，對應的x為59，則z=(3-(3-20)²/200)*(1.5-(1-1)/50)=2.33

2、設定三個指數閾值，對應三個交易量，超過閾值則執行一個下單指令，相鄰兩個同方向的訂單間隔不小於5分鐘。

import matplotlib.pyplot as plt
import pandas as pd
import MetaTrader5 as mt5
import pymt5
import time
from datetime import datetime
from datetime import timedelta
import pytz
import numpy as np
import
 
 math
import os


# 連線到MetaTrader 5
if not mt5.initialize():
	print("initialize() failed")

#登陸mt5
authorized=mt5.login(39304351, 'heniu1me', 'MetaQuotes-Demo')
if authorized:
	print("MetaTrader5 package author: ",mt5.__author__)
	print("MetaTrader5 package version: ",mt5.
 
__version__)
else:
	print("login failed")
	exit(0)

#建立當前訂單csv檔案
if not os.path.exists("current orders.csv"):
	dataframe =pd.DataFrame(columns=('Time','Type','Volume','Price','sl','tp','Note'))
	dataframe.to_csv("current orders.csv",index=False)

#建立平倉單csv檔案
if not os.path.exists("history records.csv"):
	dataframe =pd.DataFrame(columns=('Opentime','Closetime','Type','Volume','Openprice','Closeprice','sl','tp','Note'))
	dataframe.to_csv("history records.csv",index=False)


class TradingStratege(object):
	"""docstring for TradingStratege
	Args:
		symbol：交易品種
		timeframe：K線週期
		period：用於策略分析的時間序列長度(單位hours)
		target：止盈止損目標，與建倉價格的距離
		index1~index3:觸發開平倉指令的指數水平
		vol1~vol3:對應的3個交易量
		max_position:最大倉位
	"""
	def __init__(self, symbol, timeframe, period, sl_target, tp_target, index1, index2, index3, vol1, vol2, vol3, max_position):
		self.symbol = symbol
		self.timeframe = timeframe
		self.period = period
		self.sl_target = sl_target
		self.tp_target = tp_target
		self.index1 = index1
		self.index2 = index2
		self.index3 = index3
		self.vol1 = vol1
		self.vol2 = vol2
		self.vol3 = vol3
		self.max_position = max_position

	#獲取指定品種歷史資料
	def get_data(self, symbol, timeframe, period):
		mt5_timeframe = mt5.TIMEFRAME_M1
		if timeframe == "M1":
			mt5_timeframe = mt5.TIMEFRAME_M1
		elif timeframe == 'M5':
			mt5_timeframe = mt5.TIMEFRAME_M5
		elif timeframe == 'M15':
			mt5_timeframe = mt5.TIMEFRAME_M15
		elif timeframe == 'M30':
			mt5_timeframe = mt5.TIMEFRAME_M30
		elif timeframe == 'H1':
			mt5_timeframe = mt5.TIMEFRAME_H1
		elif timeframe == 'H4':
			mt5_timeframe = mt5.TIMEFRAME_H4
		elif timeframe == 'D1':
			mt5_timeframe = mt5.TIMEFRAME_D1
		# 將時區設定為GMT-2
		timezone = pytz.timezone("Etc/GMT-2")
		# 調節時間與mt5上的now相對應
		now=datetime.now(timezone)+timedelta(hours=2)
		time_from=now-timedelta(hours=period)
		# 獲取最新1小時內約60根一分鐘K線資料
		rates = mt5.copy_rates_range(symbol, mt5_timeframe, time_from, now)
		#獲取指定品種最新報價
		lasttick=mt5.symbol_info_tick(symbol)
		#將資料儲存在陣列中
		Last_price=[]
		High_price=[]
		Low_price=[]
		for price_data in rates:
			last_price=lasttick.bid
			open_price=price_data[1]
			high_price=price_data[2]	
			low_price=price_data[3]
			close_price=price_data[4]
			Last_price.append(last_price)
			High_price.append(high_price)
			Low_price.append(low_price)
		Lasttick=np.array([lasttick[0],lasttick[1],lasttick[2]])
		#返回陣列
		return Last_price,High_price,Low_price,Lasttick

	#獲取30s內最高和最低價，用於檢查是否觸發止損止盈
	def getticks(self):
		timezone = pytz.timezone("Etc/GMT-2")
		now=datetime.now(timezone)+timedelta(hours=2)
		time_from=now-timedelta(seconds=30)
		# request 100 XAUUSD ticks starting from 30s before.
		ticks = mt5.copy_ticks_from(self.symbol, time_from, 100, mt5.COPY_TICKS_ALL)
		Bid=[]
		Ask=[]
		for x in ticks:
			bid=x[1]
			ask=x[2]
			Bid.append(bid)
			Ask.append(ask)
		bid_min=min(Bid)
		bid_max=max(Bid)
		ask_min=min(Ask)
		ask_max=max(Ask)
		ticks=np.array([bid_min,bid_max,ask_min,ask_max])
		return ticks

	#計算短線多空指數
	def checkprice(self, last_price, high_price, low_price):
		#計算上漲和下跌幅度
		decrease=np.array(high_price)-np.array(last_price)
		increase=np.array(last_price)-np.array(low_price)
		#建立一個以時間距離為變數的減函式擬合出正常波動範圍
		refer=self.normal_fluctuation(increase)
		time_len1=np.arange(len(increase),0,-1)
		time_len2=np.arange(len(decrease),0,-1)
		#篩選出符合條件的資料
		filter1 = increase > refer
		filter2 = decrease > refer
		fil_inc = increase[filter1]
		fil_dec = decrease[filter2]
		fil_time1 = time_len1[filter1] 
		fil_time2 = time_len2[filter2] 
		#計算短線多空指數
		index_delta1=self.inc_warn(fil_inc,fil_time1,time_len1)
		index_delta2=self.dec_warn(fil_dec,fil_time2,time_len2)

		return index_delta1,index_delta2


	#每個10分鐘時間段選取一條提示資訊，並返回短線做空指標
	def inc_warn(self, fil_inc, fil_time, time_len):
		#判斷是否為空陣列
		if not np.size(fil_inc)==0:
			i=0
			z=int(len(time_len)/10)
			index=0
			index_delta=0
			#以10分鐘為單位對時間進行切割
			while i< z+1:
				#將時間陣列切割為z部分，得到索引陣列
				mask = np.logical_and(fil_time<=(i+1)*10,fil_time>i*10)
				j=fil_inc[mask]	
				if not np.size(j)==0:
					maxx=np.max(j)
					#用索引公式查詢時間陣列中的元素
					point=fil_time[np.where(fil_inc==maxx)][0]
					#計算短線多空指數，價格因子：(3-math.pow((maxx-20),2)/200)；時間因子：(1.5-(point-1)/50)
					index=self.market_index(maxx,point)
					print("短線快速上漲:",point,"分鐘內上漲",round(maxx,2),"美元")
					if index>index_delta:
						index_delta=index
				i=i+1
			index_delta=-1*index_delta
			return index_delta

	def dec_warn(self, fil_inc, fil_time, time_len):
		if not np.size(fil_inc)==0:
			index=0
			index_delta=0
			i=0
			z=int(len(time_len)/10)
			while i< z+1:
				mask = np.logical_and(fil_time<=(i+1)*10,fil_time>i*10)
				j=fil_inc[mask]	
				if not np.size(j)==0:
					maxx=np.max(j)
					point=fil_time[np.where(fil_inc==maxx)][0]
					index=self.market_index(maxx,point)
					print("短線快速下跌:",point,"分鐘內下跌",round(maxx,2),"美元")
					if index>index_delta:
						index_delta=index
				i=i+1
			return index_delta

	#計算當前倉位
	def current_position(self):
		a=pd.read_csv('current orders.csv')
		b=np.array(a)[:,2]
		c=b.sum()
		return c

	#根據多空指數計算開平倉數量和方向	
	def caculate(self, buy_index, sell_index):
		Type=""
		volume=0
		if buy_index !=None and sell_index== None:
			Type="buy"
			if buy_index>self.index3:
				volume=self.vol3
			elif buy_index>self.index2:
				volume=self.vol2
			elif buy_index>self.index1:
				volume=self.vol1
			else:
				volume=0
		elif sell_index != None and buy_index == None:
			Type="sell"
			if sell_index<-self.index3:
				volume=-self.vol3
			elif sell_index<-self.index2:
				volume=-self.vol2
			elif sell_index<-self.index1:
				volume=-self.vol1
			else:
				volume=0
		elif sell_index != None and buy_index != None:
			print("多空不明朗")
		else:
			pass
		
		#避免短時間內重複下單，下單間隔設定為5分鐘
		now=datetime.now()-timedelta(hours=6)
		a=pd.read_csv("current orders.csv")
		if not a.empty:
			b=a.iloc[-1,a.columns.get_loc("Time")]
			c=datetime.strptime(b,'%Y-%m-%d %H:%M:%S')
			now=datetime.now()-timedelta(hours=6)
			d=a.iloc[-1,a.columns.get_loc("Type")]
			e=now-c
			dif1=e.total_seconds()
			if volume>0 and d=="buy":
				if dif1<300:
					print("5分鐘內不再重複揸多")
					volume=0
				else:
					pass
			elif volume<0 and d=="sell":
				if  dif1<300:
					print("5分鐘內不再重複沽空")
					volume=0
				else:
					pass
			else:
				pass
		f=pd.read_csv("history records.csv")
		if not f.empty:
			g=f.iloc[-1,f.columns.get_loc("Closetime")]
			h=datetime.strptime(g,'%Y-%m-%d %H:%M:%S')
			i=f.iloc[-1,f.columns.get_loc("Type")]
			j=now-h
			dif2=j.total_seconds()
			if volume>0 and i=="sell":
				if  dif2<300:
					print("5分鐘內不再重複揸多")
					volume=0
				else:
					pass
			elif volume<0 and i=="buy":
				if dif2<300:
					print("5分鐘內不再重複沽空")
					volume=0
				else:
					pass
			else:
				pass

		return Type,volume

	def order(self, buy_index, sell_index, lasttick):
		o=pd.read_csv('current orders.csv')
		p=self.current_position()
		timezone = pytz.timezone("Etc/GMT-2")	
		time=datetime.now(timezone).strftime('%Y-%m-%d %H:%M:%S')
		Type,volume=self.caculate(buy_index,sell_index)

		if volume>0:
			if p>self.max_position:
				print("倉位已滿")
			elif p>=0:
				sl,tp=self.set_target(volume,lasttick[2])
				note="opentrade"
				print("多開")
				self.add_current_order(time,Type,volume,lasttick[2],sl,tp,note)
			elif p>(volume*(-1)):
				print("反手做多")
				for index, row in o.iterrows():
					o.loc[index,"Note"]="closetrade"
					self.add_history_records(o.loc[index,"Time"],time,o.loc[index,"Type"],o.loc[index,"Volume"],o.loc[index,"Price"],lasttick[1],o.loc[index,"sl"],o.loc[index,"tp"],o.loc[index,"Note"])
				sl,tp=self.set_target(volume,lasttick[2])
				data = pd.DataFrame({'Time':[time], 'Type': ["buy"], 'Volume': [volume+p], 'Price': [lasttick[2]],"sl":[sl],"tp":[tp],"Note":["opentrade"]}) 
				data.to_csv("current orders.csv",index=False)
			elif p==(volume*(-1)):
				print("空平")
				for index, row in o.iterrows():
					o.loc[index,"Note"]="closetrade"
					self.add_history_records(o.loc[index,"Time"],time,o.loc[index,"Type"],o.loc[index,"Volume"],o.loc[index,"Price"],lasttick[1],o.loc[index,"sl"],o.loc[index,"tp"],o.loc[index,"Note"])
				data = pd.DataFrame({'Time':[], 'Type': [], 'Volume': [], 'Price': [],"sl":[],"tp":[],"Note":[]}) 
				data.to_csv("current orders.csv",index=False)				
			elif p<(volume*(-1)):
				print("空平")
				for index, row in o.iterrows():
					x=o.loc[index,"Volume"]
					if x>-volume:
						o.loc[index,"Note"]="closetrade"
						self.add_history_records(o.loc[index,"Time"],time,o.loc[index,"Type"],o.loc[index,"Volume"],o.loc[index,"Price"],lasttick[1],o.loc[index,"sl"],o.loc[index,"tp"],o.loc[index,"Note"])
					elif x+volume==0:
						o.loc[index,"Note"]="closetrade"
						self.add_history_records(o.loc[index,"Time"],time,o.loc[index,"Type"],o.loc[index,"Volume"],o.loc[index,"Price"],lasttick[1],o.loc[index,"sl"],o.loc[index,"tp"],o.loc[index,"Note"])
						break
					elif x<-volume:
						self.add_history_records(o.loc[index,"Time"],time,o.loc[index,"Type"],-volume,o.loc[index,"Price"],lasttick[1],o.loc[index,"sl"],o.loc[index,"tp"],"closetrade")
						o.loc[index,"Volume"]=x+volume
						break
					volume=volume+x
				n=o[~o['Note'].isin(["closetrade"])]
				n.to_csv("current orders.csv",index=False)
		if volume<0:
			if p<-self.max_position:
				print("倉位已滿")
			elif p<=0:
				sl,tp=self.set_target(volume,lasttick[1])
				note="opentrade"
				print("空開")
				self.add_current_order(time,Type,volume,lasttick[1],sl,tp,note)
			elif p<(volume*(-1)):
				print("反手做空")
				for index, row in o.iterrows():
					o.loc[index,"Note"]="closetrade"
					self.add_history_records(o.loc[index,"Time"],time,o.loc[index,"Type"],o.loc[index,"Volume"],o.loc[index,"Price"],lasttick[2],o.loc[index,"sl"],o.loc[index,"tp"],o.loc[index,"Note"])
				sl,tp=self.set_target(volume,lasttick[1])
				data = pd.DataFrame({'Time':[time], 'Type': ["sell"], 'Volume': [volume+p], 'Price': [lasttick[1]],"sl":[sl],"tp":[tp],"Note":["opentrade"]}) 
				data.to_csv("current orders.csv",index=False)
			elif p==(volume*(-1)):
				print("多平")
				for index, row in o.iterrows():
					o.loc[index,"Note"]="closetrade"
					self.add_history_records(o.loc[index,"Time"],time,o.loc[index,"Type"],o.loc[index,"Volume"],o.loc[index,"Price"],lasttick[2],o.loc[index,"sl"],o.loc[index,"tp"],o.loc[index,"Note"])
				data = pd.DataFrame({'Time':[], 'Type': [], 'Volume': [], 'Price': [],"sl":[],"tp":[],"Note":[]}) 
				data.to_csv("current orders.csv",index=False)				
			elif p>(volume*(-1)):
				print("多平")
				for index, row in o.iterrows():
					x=o.loc[index,"Volume"]
					if x<-volume:
						o.loc[index,"Note"]="closetrade"
						self.add_history_records(o.loc[index,"Time"],time,o.loc[index,"Type"],o.loc[index,"Volume"],o.loc[index,"Price"],lasttick[2],o.loc[index,"sl"],o.loc[index,"tp"],o.loc[index,"Note"])
					elif x+volume==0:
						o.loc[index,"Note"]="closetrade"
						self.add_history_records(o.loc[index,"Time"],time,o.loc[index,"Type"],o.loc[index,"Volume"],o.loc[index,"Price"],lasttick[2],o.loc[index,"sl"],o.loc[index,"tp"],o.loc[index,"Note"])
						break
					elif x>-volume:
						self.add_history_records(o.loc[index,"Time"],time,o.loc[index,"Type"],-volume,o.loc[index,"Price"],lasttick[2],o.loc[index,"sl"],o.loc[index,"tp"],"closetrade")
						o.loc[index,"Volume"]=x+volume
						break
					volume=volume+x
				n=o[~o['Note'].isin(["closetrade"])]
				n.to_csv("current orders.csv",index=False)
		else:
			pass

	#指數提醒功能
	def index_warn(self, index_dec, index_inc):
		if not index_dec==None:
			print("短線做空指數：",round(index_dec,2))
		if not index_inc==None:
			print("短線做多指數：",round(index_inc,2))

	#增加訂單資料
	def add_current_order(self, time, Type, volume, price, sl, tp, note):
		file = os.getcwd() + '\\current orders.csv'  #指明檔案位置，即當前工作路徑下的csv檔案
		data = pd.DataFrame({'Time':[time], 'Type': [Type], 'Volume': [volume], 'Price': [price], "sl":[sl], "tp":[tp], "Note":[note]}) 
		data.to_csv(file, index=False, mode='a+', header=False)

	#增加歷史資料
	def add_history_records(self,opentime,closetime,Type,volume,openprice,closeprice,sl,tp,note):
		file = os.getcwd() + '\\history records.csv'  #指明檔案位置，即當前工作路徑下的csv檔案
		data = pd.DataFrame({'Opentime':[opentime], 'Closetime':[closetime], 'Type': [Type], 'Volume': [volume], 'Openprice': [openprice], 'Closeprice':[closeprice], "sl":[sl], "tp":[tp], "Note":[note]}) 
		data.to_csv(file, index=False, mode='a+', header=False)

	#設定止盈止損
	def set_target(self, volume, price):
		sl=0
		tp=0
		if volume>0:
			sl=price-self.sl_target
			tp=price+self.tp_target
		else:
			sl=price+self.sl_target
			tp=price-self.tp_target
		return sl,tp

	#檢查當前訂單是否觸發止盈止損
	def check_orders(self):
		p=pd.read_csv('current orders.csv')
		last=self.getticks()
		timezone = pytz.timezone("Etc/GMT-2")
		closetime=datetime.now(timezone).strftime('%Y-%m-%d %H:%M:%S')
		for index, row in p.iterrows():
			if  p.loc[index,"Type"]=="buy":
				if p.loc[index,"sl"]>last[0]:
					p.loc[index,"Note"]="stoploss"
					print("觸發止損")
					self.add_history_records(p.loc[index,"Time"],closetime,p.loc[index,"Type"],p.loc[index,"Volume"],p.loc[index,"Price"],p.loc[index,"sl"],p.loc[index,"sl"],p.loc[index,"tp"],p.loc[index,"Note"])
				elif p.loc[index,"tp"]<last[1]:
					p.loc[index,"Note"]="takeprofit"
					print("觸發止盈")						
					self.add_history_records(p.loc[index,"Time"],closetime,p.loc[index,"Type"],p.loc[index,"Volume"],p.loc[index,"Price"],p.loc[index,"tp"],p.loc[index,"sl"],p.loc[index,"tp"],p.loc[index,"Note"])
				else:
					pass
			elif p.loc[index,"Type"]=="sell":
				if p.loc[index,"sl"]<last[3]:
					p.loc[index,"Note"]="stoploss"
					print("觸發止損")
					self.add_history_records(p.loc[index,"Time"],closetime,p.loc[index,"Type"],p.loc[index,"Volume"],p.loc[index,"Price"],p.loc[index,"sl"],p.loc[index,"sl"],p.loc[index,"tp"],p.loc[index,"Note"])
				elif p.loc[index,"tp"]>last[2]:
					p.loc[index,"Note"]="takeprofit"
					print("觸發止盈")
					self.add_history_records(p.loc[index,"Time"],closetime,p.loc[index,"Type"],p.loc[index,"Volume"],p.loc[index,"Price"],p.loc[index,"tp"],p.loc[index,"sl"],p.loc[index,"tp"],p.loc[index,"Note"])
				else:
					pass
		o=p[~p['Note'].isin(["takeprofit"])]
		o.to_csv("current orders.csv",index=False)
		p=pd.read_csv('current orders.csv')
		q=p[~p['Note'].isin(["stoploss"])]
		q.to_csv("current orders.csv",index=False)

	def running(self):
		while 1 == 1:
			Last_price,High_price,Low_price,Lasttick=self.get_data(self.symbol,self.timeframe,self.period)
			index_dec,index_inc=self.checkprice(Last_price,High_price,Low_price)
			self.index_warn(index_dec,index_inc)
			self.order(index_inc,index_dec,Lasttick)
			self.check_orders()
			print("------")
			time.sleep(10)


class XAUUSD_Stratege(TradingStratege):
	def __init__(self, symbol, timeframe, period, sl_target, tp_target, index1, index2, index3, vol1, vol2, vol3, max_position):
		TradingStratege.__init__(self, symbol, timeframe, period, sl_target, tp_target, index1, index2, index3, vol1, vol2, vol3, max_position)
	'''
	構造一個以時間為變數的XAUUSD正常波動範圍函式：
		1分鐘內的正常波動範圍為0~2.5，
		5分鐘內的正常波動範圍為0~4，
		60分鐘內的正常波動範圍為0~10，
		隨著時間的延長，價格波動範圍擴大，但增速呈遞減趨勢。
		超過正常波動範圍則發出提醒。
	'''
	def normal_fluctuation(self, array):
		refer=np.array([10*math.sin(pow((x/(50*self.period)),1.5)+0.5*math.pi) for x in range(1,len(array)+1)])
		return refer
	#構建一個XAUUSD多空指數模型，價格因子：(3-math.pow((maxx-20),2)/200)；時間因子：(1.5-(point-1)/50)，
	#理論上越短時間內波幅越大，指數越高。
	def market_index(self, maxx, time):
		index=(3-math.pow((maxx-20),2)/200)*(1.5-(time-1)/50)
		return index


if __name__ == '__main__':
	xauusd_trading=XAUUSD_Stratege('XAUUSD','M1',1,10,15,2.5,2.8,3.2,0.2,0.4,0.7,3)
	xauusd_trading.running()