网格策略
由bqazbxz1创建，最终由bqazbxz1 被浏览 2 用户
from collections import deque
import pandas as pd
import math
def initialize(context):
# 策略参数
g.security = "513310.SS" # 替换为您的标的
g.trade_value = 2500 # 每次交易金额(元)
g.buy_threshold = 0.05 # 5%
g.sell_threshold = 0.1 # 10%
g.file_path = get_research_path() + '/order_queue.csv'
g.order_queue = load_order_queue()
set_universe(g.security)
def buy(current_price):
trade_amount = calculate_lots(g.trade_value,current_price)
#order_id = order(g.security, trade_amount)
buy_order = {}
# buy_order["OrderID"] = order_id
buy_order["SecurityID"] = g.security
buy_order["TradePrice"] = current_price
# buy_order["TradeTime"] = datetime.now().strftime("%Y-%m-%d %H-%M")
buy_order["TradeVolume"] = trade_amount
log.info(f"Order Buying {g.security} cash {g.trade_value} Successed")
g.order_queue = pd.concat([g.order_queue,pd.DataFrame([buy_order])])
save_order_queue(g.order_queue)
def sell(trade_amount):
# order(g.security,-trade_amount)
g.order_queue = g.order_queue.iloc[:-1]
save_order_queue(g.order_queue)
log.info(f"Order Selling {g.security} amount {trade_amount} Successed")
def handle_data(context, data):
current_price = data[g.security].close
if len(g.order_queue) == 0:
buy(current_price)
return
# 获取最后一笔买入订单
if len(g.order_queue) == 0 :
log.info(f"Check Order Selling {g.security} Failed For empty order queue")
return
last_order_amount = g.order_queue['TradeVolume'].iloc[-1]
last_order_price = g.order_queue['TradePrice'].iloc[-1]
log.info(f"current_price {current_price}")
log.info(f"last_price {last_order_price}")
log.info(f"upper_price {last_order_price * (1 + g.sell_threshold)}")
log.info(f"lower_price {last_order_price * (1 - g.buy_threshold)}")
# 检查卖出条件(涨幅超过1%)
if current_price >= last_order_price * (1 + g.sell_threshold):
sell(last_order_amount)
# 检查买入条件(跌幅超过1%)
elif current_price <= last_order_price * (1 - g.buy_threshold):
buy(current_price)
def load_order_queue():
try:
print(g.file_path)
order_df = pd.read_csv(g.file_path)
except FileNotFoundError:
log.error(f'order file not found')
return None
except Exception as e:
log.error(f'read order file error {e}')
return pd.DataFrame(columns=["OrderID","SecurityID","TradePrice","TradeVolume","TradeTime"])
if len(order_df) == 0:
return pd.DataFrame(columns=["OrderID","SecurityID","TradePrice","TradeVolume","TradeTime"])
return order_df
def save_order_queue(order_df):
try:
order_df.to_csv(g.file_path,index=False)
except Exception as e:
log.error(f'save order file error {e}')
def calculate_lots(cash, price):
if price <= 0:
raise ValueError("价格必须大于0")
if cash <= 0:
return 0
max_shares = cash // price
lots = math.floor(max_shares / 100)
return max(lots, 0)*100
\