等级: 新手上路
- 注册:
- 2021-7-30
- 曾用名:
|
楼主 |
发表于 2025-7-23 11:50
|
显示全部楼层
# 本Python代码主要用于策略交易
# 可以自己import我们平台支持的第三方python模块，比如pandas、numpy等。
from PythonApi import *
import datetime
import re # 导入正则表达式模块
import numpy as np
from datetime import datetime # ← 直接导入类
# 在这个方法中编写任何的初始化逻辑。context对象将会在算法策略运行时使用
def init(context):
# 在context中保存全局变量
block = get_blocks('自选股', 1) # ['DQLH09','DQP09']
context.universe = block # 将交易的合约设置为连续合约
context.account_id = None # 初始时设置为None
# 设置定时器，每2秒执行一次GridTrade函数
settimer(GridTrade,2000) # 2秒执行一次
# 初始化账户信息
update_account_info(context)
# GridTrade函数（每2秒执行一次）
#退出程序后杀死计时器
def exit(context):
killtimer(GridTrade)
def GridTrade(context):
write_logging("=== GridTrade 触发 ===")
buy_open_long2(context)
portfolio_query_module(context)
def buy_open_long2(context):
codes = context.universe
write_logging(f"当前股票池: {codes}")
if not codes:
write_logging("股票池为空，跳过")
return
for code in codes:
stock_code_num = re.sub(r'\D', '', code)
longtrades = stock_code_num + "_longstoday"
write_logging(f"检查信号: {longtrades}")
signal = getextdata(longtrades)
write_logging(f"信号值: {signal}")
if signal == 1:
write_logging(f"满足开仓条件，准备买入: {code}")
buy_open_module(context, code, len(codes))
setextdata(longtrades, 2)
write_logging("已设置信号为 2")
else:
write_logging(f"不满足开仓条件，跳过: {code}")
def buy_open_module(context, code, blocks):
write_logging("=== buy_open_module start ===")
write_logging(f"入参 code={code} blocks={blocks}")
# 1) 生成信号 key
stock_code_num = re.sub(r'\D', '', code)
longtrades_key = stock_code_num + "_longstoday"
write_logging(f"longtrades_key = {longtrades_key}")
# 2) 读取资金数据
available_funds = getextdata('available_funds')
dynastic_rights = getextdata('dynastic_rights')
write_logging(f"available_funds = {available_funds}")
write_logging(f"dynastic_rights = {dynastic_rights}")
# 3) 资金比例判断
if blocks == 0:
write_logging("blocks == 0，直接退出")
return
if dynastic_rights is None or dynastic_rights == 0:
write_logging("dynastic_rights 为 None 或 0，无法计算比例，退出")
return
if available_funds is None:
write_logging("available_funds 为 None，退出")
return
ratio = available_funds / dynastic_rights
write_logging(f"资金比例 = {ratio:.4f}")
if ratio <= 0.2:
write_logging("资金比例 <= 0.2，不满足开仓条件")
return
# 4) 计算每只股票可分配金额
value = available_funds / max(3, blocks)
write_logging(f"每只股票可分配金额 value = {value}")
# 5) 获取最新价
current_price = np.round(get_dynainf(code, 7), 2)
write_logging(f"最新价 current_price = {current_price}")
if current_price is None or current_price <= 0:
write_logging(f"未获取到 {code} 的有效最新价，退出")
return
# 6) 获取涨停价（字段 54）
limit_up = np.round(get_dynainf(code, 54), 2)
write_logging(f"涨停价 limit_up = {limit_up}")
if limit_up is None or limit_up <= 0:
write_logging("涨停价无效，退出")
return
# 7) 检查最小买入金额
min_buy = 100 * current_price
write_logging(f"最小买入金额(100 股) = {min_buy}")
if value < min_buy:
write_logging("可分配金额不足以买入 100 股，退出")
return
# 8) 价格条件判断
write_logging(f"价格判断: current_price {current_price} <= limit_up {limit_up}")
if current_price <= limit_up:
write_logging("满足开仓条件，准备下单")
# 检查你的代码中所有 buy_open 调用，统一改为：
# 按金额买入（volume=0）
#市价委托方式，买入平安银行1000股
buy_open(code, "Market", 0 , value,serial_id = 1)
write_logging(f"下单返回 order_id = {order_id}")
setextdata(longtrades_key, 2)
write_logging(f"已设置 {longtrades_key} = 2")
else:
write_logging("最新价高于涨停价，不开仓")
write_logging("=== buy_open_module end ===")
# 打印账户中的股票持仓信息
## 示例用法
def handle_bar(context):
# 更新账户信息
update_account_info(context)
# 这里可以添加根据持仓和账户信息进行交易逻辑的代码
#def buy_open_long(context):
# codes = context.universe
# # 开仓前持仓全集变量
#
# if len(codes) == 0:
# return
#
# for code in codes:
# # 提取股票代码中的数字部分
# stock_code_num = re.sub(r'\D', '', code)
# longtrades = stock_code_num + "_longstoday"
#
## if not istradertime(code):
## continue
# if getextdata('longtrades')==1:
# # 示例开仓
# buy_open_module(context, code, len(codes) )
# setextdata('longtrades',2)
def buy_open_long(context):
codes = context.universe
print("当前股票池 codes:", codes)
if len(codes) == 0:
print("股票池为空，跳过开仓逻辑")
return
for code in codes:
print(f"处理股票: {code}")
# 提取股票代码中的数字部分
stock_code_num = re.sub(r'\D', '', code)
longtrades = stock_code_num + "_longstoday"
print(f"生成的 longtrades 键名: {longtrades}")
# 获取 longtrades 的扩展数据
ext_value = getextdata(longtrades)
print(f"getextdata('{longtrades}') 的值为: {ext_value}")
if ext_value == 1:
print(f"条件满足，准备开仓: {code}")
buy_open_module(context, code, len(codes))
setextdata(longtrades, 2)
print(f"已设置 {longtrades} 为 2")
else:
print(f"条件不满足，跳过开仓: {code}，值为 {ext_value}")
# BUY开仓模块
#def buy_open_module(context, code ,blocks):
# # 保存全局变量longtrades，表示开仓前的持仓情况
# # 开仓前持仓全集变量
# # 提取股票代码中的数字部分
# stock_code_num = re.sub(r'\D', '', code)
# longtrades = stock_code_num + "_longstoday"
# #setextdata('longtrades', volume)
## dynastic_rights = getextdata('dynastic_rights')
## available_funds = getextdata('available_funds')
# if blocks != 0 and getextdata('available_funds')/ getextdata('dynastic_rights') > 0.2:
# # Calculate the value per security (though we'll only use it for the first three)
# value = getextdata('available_funds')/ max(3, blocks)
# # 获取最新价格
# #获取rb00当前最新价
# #get_dynainf('SQRB00', 7)
# current_price = np.round(get_dynainf(code, 7),2)
# if current_price is None:
# write_logging(f"未获取到 {code} 的最新价格。")
# # Calculate the ratio for all qualified stocks
# # Check if we can buy at least 100 shares
# if value / get_dynainf(code, 54) > 100:
# # Determine the order value based on the ratio
# if current_price <= np.round(get_dynainf(code, 54),2):#价格54是涨停
# # 执行开仓操作
# # 使用账号'12345'以市价方式买入价值20000元的螺纹钢
# buy_open(code, "Market", amount=value,serial_id = 1)
# #order_id = buy_open(code, 'Limit', current_price, value,serial_id = 1)# 控制每次下单金额为value
# setextdata('longtrades',2)
## if current_price <= np.round(min(get_dynainf(code, 28)*1.02,get_dynainf(code, 54)),2):#49是买五价,54是涨停,51卖五价
## # 执行开仓操作
## # 使用账号'12345'以市价方式买入价值20000元的螺纹钢
## #buy_open(code, "Market", amount=value,serial_id = 1)
## #buy_open(code, 'Limit', np.round(min(get_dynainf(code, 28)*1.02,get_dynainf(code, 54)),2), value,serial_id = 2)# 控制每次下单金额为value
## setextdata('longtrades',2)
# SELL平仓模块
def sellclose_module(context, code, price, volume):
try:
# 获取全局变量stopline，表示持仓信息平仓变量
# ---------- 当前时间 ----------
now = datetime.now()
current_time = now.time()
#write_logging(f"当前时间: {current_time}")
#current_time = datetime.datetime.now().time()
# 提取小时和分钟
current_hour = current_time.hour
current_minute = current_time.minute
#write_logging(f"当前时间: {current_hour}:{current_minute}")
# 提取股票代码中的数字部分
stock_code_num = re.sub(r'\D', '', code)
stopline_key = f"{stock_code_num}_stopline"
# 如已标记平仓则直接跳过
if int(getextdata(stopline_key) or 0) == 1:
write_logging(f"{code} 已平仓，跳过")
return None
# 获取最新价格
#get_dynainf('SQRB00', 7)
current_price = np.round(get_dynainf(code, 7),2)
Lower_Limitprice=np.round(get_dynainf(code, 55),2)#55,跌停，请用英文
if current_price is None:
write_logging(f"未获取到 {code} 的最新价格。")
return
# 读取成本价格
cost_price = getextdata(f'{stock_code_num}_cost')
if cost_price is None:
write_logging(f"未获取到 {code} 的成本价格。")
return
# 获取五日均线价格
close_data = history_bars(code, 5, '1d', 'CLOSE')
if close_data is not None :
ma5 = np.round(np.mean(close_data),2)
#write_logging(f"五日均线价格:{ma5}: {code}:{price}:{volume}:{current_price}")
# 上午10点：亏损5%时平仓,get_dynainf函数55跌停
if current_time.hour == 10 and current_time.minute == 0:
if (cost_price - current_price) / cost_price >= 0.05:
sell_close(code, 'Limit', Lower_Limitprice, volume,serial_id = 2)
setextdata(stopline_key, 1) # ← 关键：标记已平仓
write_logging(f"亏损5%平仓: {code}, 成本价格: {cost_price}, 当前价格: {current_price}")
return order_id
# 上午11点：当前最新价格低于昨日收盘价时平仓
elif current_time.hour == 11 and current_time.minute == 0:
# 获取昨日收盘价
yesterday_close = get_dynainf(code, 3)
if current_price < yesterday_close:
order_id = sell_close(code, 'Limit', Lower_Limitprice, volume,serial_id = 3)
setextdata(stopline_key, 1) # ← 关键：标记已平仓
write_logging(f"低于昨日收盘价平仓: {code}, 昨日收盘价: {yesterday_close}, 当前价格: {current_price}")
return order_id
# 下午2点50分：当前最新价格低于五日均线价格时平仓
elif current_time.hour == 14 and current_time.minute == 26:
stock_code_num = re.sub(r'\D', '', code)
stopline_key = stock_code_num + "_stopline"
stopline = int(getextdata(stopline_key) or 0)
if stopline == 1:
write_logging(f"{code} {stopline_key}=1，已平仓，跳过")
return None
else:
#order_id = None
if current_price < ma5:
order_id = sell_close(code, 'Limit', Lower_Limitprice, volume,serial_id = 4)
setextdata(stopline_key, 1) # ← 关键：标记已平仓
write_logging(f"低于五日均线平仓: {code}, 五日均线: {ma5}, 当前价格: {current_price}")
return order_id
except Exception as e:
write_logging(f"执行平仓模块时发生错误: {e}")
return None
# 持仓查询模块
def portfolio_query_module(context):
try:
# 确保account_id已初始化
if not hasattr(context, 'account_id') or context.account_id is None:
write_logging("账户ID未初始化，无法查询持仓信息。")
return
# 获取账户中的所有持仓品种
account_stocks = get_portfolio_book(2, context.account_id)
if account_stocks:
#write_logging("账户中的股票持仓:")
for stock in account_stocks:
# 提取股票代码中的数字部分
stock_code_num = re.sub(r'\D', '', stock)
#write_logging(f"股票代码: {stock}, 数字部分: {stock_code_num}")
# 查询该股票的持仓详情
try:
portfolio = get_portfolio(stock, 2, context.account_id)
if portfolio:
total_position = portfolio.buy_quantity + portfolio.sell_quantity
available_position = portfolio.buy_today_quantity + portfolio.sell_today_quantity
cost_price = portfolio.buy_avg_holding_price
#write_logging(f" 总持仓: {total_position}, 今日可用持仓: {available_position}, 持仓成本: {cost_price}")
# 将持仓信息保存到全局变量，格式为 "s_数字部分_信息类型"
setextdata(f'{stock_code_num}_total', total_position)
setextdata(f'{stock_code_num}_avail', available_position)
setextdata(f'{stock_code_num}_cost', cost_price)
# 示例平仓
sellclose_module(context, stock, cost_price, available_position)
else:
write_logging(f" 未获取到 {stock} 的持仓详情。")
except Exception as e:
write_logging(f"查询 {stock} 的持仓详情时发生错误: {e}")
else:
write_logging("账户中没有股票持仓。")
except Exception as e:
write_logging(f"获取账户持仓品种时发生错误: {e}")
# 更新账户信息函数
# 假设平台定义如下（需替换为你的实际参数）
def update_account_info(context):
try:
# 获取当前登录账户列表
accounts = get_account_book()
if not accounts:
write_logging("未检测到任何登录账户")
return
account_no = accounts[0] # 默认取第一个登录账户
write_logging(f"使用账户: {account_no}")
# 判断账户是否有效
if not isaccount(account_no):
write_logging("账户未登录或无效")
return
# 获取账户信息
cash_balance = get_account(3, account_no)
floating_profit = get_account(4, account_no)
dynastic_rights = get_account(6, account_no)
available_funds = get_account(19, account_no)
# 保存到扩展数据
setextdata('cash_balance', cash_balance or 0)
setextdata('floating_profit', floating_profit or 0)
setextdata('dynastic_rights', dynastic_rights or 0)
setextdata('available_funds', available_funds or 0)
write_logging(f"账户信息更新成功 - 现金余额: {cash_balance}, 可用资金: {available_funds}")
except Exception as e:
write_logging(f"更新账户信息失败: {e}")
# 测试代码
|
|
发表于 2025-7-23 11:49
发表于 2025-7-23 14:08