"""
|
掘金
|
"""
|
|
from __future__ import print_function, absolute_import
|
|
import datetime
|
import json
|
import logging
|
import time as t
|
|
import numpy
|
import schedule
|
|
import gm.api as gmapi
|
|
import big_money_num_manager
|
import client_manager
|
import code_volumn_manager
|
import constant
|
import global_data_loader
|
import global_util
|
import gpcode_first_screen_manager
|
import gpcode_manager
|
import threading
|
|
import server
|
import tool
|
|
from db import redis_manager
|
import authority
|
import decimal
|
|
from trade import trade_gui, l2_trade_util
|
from l2.cancel_buy_strategy import L2LimitUpSellStatisticUtil
|
from l2_code_operate import L2CodeOperate
|
from log import logger_juejin_tick, logger_system
|
from trade.trade_data_manager import CodeActualPriceProcessor
|
from trade.trade_queue_manager import JueJinBuy1VolumnManager
|
|
redisManager = redis_manager.RedisManager(0)
|
__jueJinBuy1VolumnManager = JueJinBuy1VolumnManager()
|
__actualPriceProcessor = CodeActualPriceProcessor()
|
|
|
# 设置账户信息
|
def setAccountInfo(accountId, strategyId, token):
|
redis = redisManager.getRedis()
|
redis.set("juejin-account-id", accountId)
|
redis.set("juejin-strategy-id", strategyId)
|
redis.set("juejin-token", token)
|
|
|
def getAccountInfo():
|
redis = redisManager.getRedis()
|
account_id = redis.get("juejin-account-id")
|
strategy_id = redis.get("juejin-strategy-id")
|
token = redis.get("juejin-token")
|
return account_id, strategy_id, token
|
|
|
def init_data():
|
# 删除所有的涨停卖数据
|
L2LimitUpSellStatisticUtil.clear()
|
# 重置所有的大单数据
|
big_money_num_manager.reset_all()
|
# 清除水下捞数据
|
__actualPriceProcessor.clear_under_water_data()
|
# 载入行业股票代码
|
global_data_loader.load_industry()
|
# 载入代码自由流通市值
|
global_data_loader.load_zyltgb()
|
# 载入量
|
global_data_loader.load_volumn()
|
|
# 9点25之前删除所有代码
|
if tool.trade_time_sub(tool.get_now_time_str(), "09:25:00") <= 0:
|
# 删除L2监听代码
|
gpcode_manager.clear_listen_codes()
|
# 删除首板代码
|
gpcode_manager.clear_first_codes()
|
# 删除首板未筛选代码
|
gpcode_first_screen_manager.clear_first_no_screen_codes()
|
# 删除禁止代码
|
l2_trade_util.init_forbidden_trade_codes()
|
# 清空白名单
|
l2_trade_util.WhiteListCodeManager.clear()
|
# 清空想要买
|
gpcode_manager.WantBuyCodesManager.clear()
|
|
|
# 每日初始化
|
def everyday_init():
|
# 交易時間不能做初始化
|
if not tool.is_init_time():
|
raise Exception("交易时间不能初始化")
|
|
codes = gpcode_manager.get_gp_list()
|
logger_system.info("每日初始化")
|
|
# 今日实时涨停
|
global_data_loader.add_limit_up_codes([], True)
|
# 主要获取收盘价
|
get_latest_info(None)
|
# 获取60天最大量与昨日量
|
global_util.today_volumn.clear()
|
global_util.max60_volumn.clear()
|
global_util.yesterday_volumn.clear()
|
volumn_dict = get_volumns(codes)
|
for key in volumn_dict:
|
code_volumn_manager.set_histry_volumn(key, volumn_dict[key][0], volumn_dict[key][1])
|
# 清除大单数据
|
global_util.big_money_num.clear()
|
# 初始化大单数据
|
for code in codes:
|
big_money_num_manager.add_num(code, 0)
|
big_money_num_manager.expire(code)
|
# 清除涨停时间
|
global_util.limit_up_time.clear()
|
init_data()
|
# 初始化同花顺主站
|
l2_clients = client_manager.getValidL2Clients()
|
for client in l2_clients:
|
try:
|
server.repair_ths_main_site(client)
|
except Exception as e:
|
pass
|
|
|
def __run_schedule():
|
while True:
|
schedule.run_pending()
|
|
|
def init(context):
|
# gmapi.subscribe(symbols="SZSE.002529", frequency="1d", count=30)
|
|
# 订阅浦发银行, bar频率为一天和一分钟
|
# 订阅订阅多个频率的数据，可多次调用subscribe
|
# 获取需要监听的股票
|
init_data()
|
logger_system.info("掘金初始化")
|
schedule.every().day.at("09:15:00").do(everyday_init)
|
t1 = threading.Thread(target=lambda: __run_schedule())
|
# 后台运行
|
t1.setDaemon(True)
|
t1.start()
|
|
# 多个时间点获取收盘价
|
gmapi.schedule(schedule_func=get_latest_info, date_rule='1d', time_rule='08:30:00')
|
gmapi.schedule(schedule_func=get_latest_info, date_rule='1d', time_rule='08:50:00')
|
gmapi.schedule(schedule_func=get_latest_info, date_rule='1d', time_rule='09:28:00')
|
gmapi.schedule(schedule_func=get_current_info, date_rule='1d', time_rule='09:25:00')
|
gmapi.schedule(schedule_func=get_current_info, date_rule='1d', time_rule='09:29:00')
|
re_subscribe_tick()
|
# re_subscribe_bar()
|
|
# 初始化内容
|
clients = authority.get_l2_clients()
|
for client in clients:
|
for i in range(0, constant.L2_CODE_COUNT_PER_DEVICE):
|
gpcode_manager.init_listen_code_by_pos(client, i)
|
|
|
def get_latest_info(context):
|
# 初始化内容
|
clients = authority.get_l2_clients()
|
for c in clients:
|
for i in range(0, constant.L2_CODE_COUNT_PER_DEVICE):
|
gpcode_manager.init_listen_code_by_pos(int(c), i)
|
codes = gpcode_manager.get_gp_list()
|
result = JueJinManager.get_gp_latest_info(codes)
|
for item in result:
|
sec_level = item['sec_level']
|
symbol = item['symbol']
|
symbol = symbol.split(".")[1]
|
pre_close = tool.to_price(decimal.Decimal(str(item['pre_close'])))
|
if sec_level == 1:
|
if symbol in codes:
|
gpcode_manager.set_price_pre(symbol, pre_close)
|
else:
|
gpcode_manager.rm_gp(symbol)
|
|
|
# 获取最新的信息
|
def get_current_info():
|
data = gpcode_manager.get_gp_list()
|
results = JueJinManager.get_gp_current_info(data)
|
logger_juejin_tick.debug("定时获取：{}", results)
|
for result in results:
|
price = result["price"]
|
symbol = result['symbol']
|
# 保存最新价
|
symbol = symbol.split(".")[1]
|
accpt_price(symbol, price)
|
|
|
# 设置收盘价
|
def re_set_price_pre(code):
|
codes = [code]
|
re_set_price_pres(codes)
|
|
|
def re_set_price_pres(codes):
|
result = JueJinManager.get_gp_latest_info(codes)
|
for item in result:
|
symbol = item['symbol']
|
symbol = symbol.split(".")[1]
|
pre_close = tool.to_price(decimal.Decimal(str(item['pre_close'])))
|
gpcode_manager.set_price_pre(symbol, pre_close)
|
|
|
__prices_now = {}
|
|
|
def on_tick(context, tick):
|
# print(tick["created_at"])
|
relative_timestamp = t.time() % (24 * 60 * 60) + 8 * 60 * 60
|
# 9点20-15:05接受数据
|
start1 = 60 * 60 * 9 + 31 * 60
|
end1 = 60 * 60 * 11 + 35 * 60
|
start2 = 60 * 60 * 12 + 50 * 60
|
end2 = 60 * 60 * 15 + 5 * 60
|
# TODO 测试
|
if (start1 < relative_timestamp < end1 or start2 < relative_timestamp < end2) or constant.TEST:
|
symbol = tick['symbol']
|
price = tick['price']
|
# print(symbol,price)
|
# 价格有变化的时候才发送操作
|
# if symbol in __prices_now and __prices_now[symbol] == price:
|
# return
|
|
# 保存最新价
|
symbol = symbol.split(".")[1]
|
JueJinManager.add_listen_code(symbol)
|
time_ = tick["created_at"].strftime("%H:%M:%S")
|
data_ = (symbol, time_, tick["quotes"][0]["bid_v"], tick["quotes"][0]["bid_p"])
|
logger_juejin_tick.info("买1量 {},{},{},{}", data_[1], data_[0], data_[2],
|
data_[3])
|
# 暂时不采用
|
# need_sync = __jueJinBuy1VolumnManager.save(data_[0], data_[1], data_[2], data_[3])
|
# if need_sync:
|
# # 同步数据
|
# L2LimitUpMoneyStatisticUtil.verify_num(data_[0], data_[2], data_[1])
|
|
# print(tick["created_at"],tick["quotes"][0]["bid_v"])
|
|
accpt_price(symbol, price)
|
__prices_now[symbol] = price
|
|
|
# 获取到现价
|
def accpt_price(code, price, price_from="juejin"):
|
return
|
gpcode_manager.set_price(code, price)
|
# 获取收盘价
|
pricePre = gpcode_manager.get_price_pre(code)
|
if pricePre is not None:
|
rate = round((price - pricePre) * 100 / pricePre, 1)
|
if rate >= 7:
|
logger_juejin_tick.info("{}-{}-{}", code, price, rate)
|
if not gpcode_manager.is_listen_old(code) and not gpcode_manager.is_operate(
|
code) and not gpcode_manager.is_listen_full():
|
L2CodeOperate.get_instance().add_operate(1, code, "现价变化,rate-{} from-{}".format(rate, price_from))
|
# 进入监控
|
elif rate < 5:
|
# 移除监控
|
if gpcode_manager.is_listen_old(code) and not gpcode_manager.is_operate(code):
|
L2CodeOperate.get_instance().add_operate(0, code, "现价变化,rate-{} from-{}".format(rate, price_from))
|
|
|
# 获取到现价
|
def accept_prices(prices):
|
# 获取首板代码
|
first_codes = gpcode_manager.get_first_gp_codes()
|
|
print("价格代码数量：", len(prices))
|
|
__actualPriceProcessor.save_current_price_codes_count(len(prices))
|
# 采集的代码数量不对
|
if len(gpcode_manager.get_gp_list()) - len(prices) > 10:
|
return
|
now_str = tool.get_now_time_str()
|
now_strs = now_str.split(":")
|
now_second = int(now_strs[0]) * 60 * 60 + int(now_strs[1]) * 60 + int(now_strs[2])
|
start = 60 * 60 * 9 + 31 * 60
|
if False:
|
for d in prices:
|
code, price = d["code"], float(d["price"])
|
accpt_price(code, price, "ths")
|
else:
|
_code_list = []
|
_delete_list = []
|
for d in prices:
|
code, price = d["code"], float(d["price"])
|
gpcode_manager.set_price(code, price)
|
# 获取收盘价
|
pricePre = gpcode_manager.get_price_pre(code)
|
if pricePre is not None:
|
rate = round((price - pricePre) * 100 / pricePre, 2)
|
if first_codes and code in first_codes:
|
rate = rate / 2
|
if rate >= 0:
|
# 暂存涨幅为正的代码
|
_code_list.append((rate, code))
|
else:
|
# 暂存涨幅为负的代码
|
_delete_list.append((rate, code))
|
try:
|
__actualPriceProcessor.process_rate(code, rate, now_str)
|
except Exception as e:
|
logging.exception(e)
|
|
try:
|
__actualPriceProcessor.save_current_price(code, price,
|
gpcode_manager.get_limit_up_price_by_preprice(
|
pricePre) == tool.to_price(
|
decimal.Decimal(d["price"])))
|
except Exception as e:
|
logging.exception(e)
|
|
# -------------------------------处理交易位置分配---------------------------------
|
# 排序
|
new_code_list = sorted(_code_list, key=lambda e: (e.__getitem__(0), e.__getitem__(1)), reverse=True)
|
# 预填充下单代码
|
_buy_win_codes = []
|
for d in new_code_list:
|
_buy_win_codes.append(d[1])
|
for d in _delete_list:
|
_buy_win_codes.append(d[1])
|
try:
|
trade_gui.THSBuyWinManagerNew.fill_codes(_buy_win_codes)
|
except Exception as e:
|
logging.exception(e)
|
pass
|
|
# -------------------------------处理L2监听---------------------------------
|
|
client_ids = client_manager.getValidL2Clients()
|
# 最多填充的代码数量
|
max_count = len(client_ids) * constant.L2_CODE_COUNT_PER_DEVICE
|
if max_count == 0:
|
max_count = constant.L2_CODE_COUNT_PER_DEVICE
|
|
_delete_list = []
|
for item in new_code_list:
|
if l2_trade_util.is_in_forbidden_trade_codes(item[1]) or item[0] < 0:
|
_delete_list.append(item)
|
|
for item in _delete_list:
|
new_code_list.remove(item)
|
# 截取前几个代码填充
|
add_list = new_code_list[:max_count]
|
# 后面的代码全部删除
|
_delete_list.extend(new_code_list[max_count:])
|
|
add_code_list = []
|
del_code_list = []
|
for d in add_list:
|
add_code_list.append(d[1])
|
|
for d in _delete_list:
|
del_code_list.append(d[1])
|
|
# 后面的代码数量
|
# 先删除应该删除的代码
|
for code in del_code_list:
|
if gpcode_manager.is_listen_old(code):
|
# 判断是否在监听里面
|
L2CodeOperate.get_instance().add_operate(0, code, "现价变化")
|
# 增加应该增加的代码
|
for code in add_code_list:
|
if not gpcode_manager.is_listen_old(code):
|
L2CodeOperate.get_instance().add_operate(1, code, "现价变化")
|
|
# 获取卡位数量
|
free_count = gpcode_manager.get_free_listen_pos_count()
|
if free_count < 2:
|
# 空闲位置不足
|
listen_codes = gpcode_manager.get_listen_codes()
|
for code in listen_codes:
|
if not gpcode_manager.is_in_gp_pool(code):
|
client_id, pos = gpcode_manager.get_listen_code_pos(code)
|
gpcode_manager.set_listen_code_by_pos(client_id, pos, "")
|
free_count += 1
|
if free_count > 2:
|
break
|
|
print(add_code_list, del_code_list)
|
|
|
def on_bar(context, bars):
|
print("on_bar", bars)
|
|
|
def re_subscribe_tick():
|
gpcode_list = gpcode_manager.get_gp_list_with_prefix()
|
codes_str = ",".join(gpcode_list)
|
print("订阅：", codes_str)
|
# TODO
|
gmapi.subscribe(symbols=codes_str, frequency='tick', count=2, wait_group=False, wait_group_timeout='10s',
|
unsubscribe_previous=True)
|
|
|
def re_subscribe_bar():
|
gpcode_list = gpcode_manager.get_gp_list_with_prefix()
|
codes_str = ",".join(gpcode_list)
|
print("订阅：", codes_str)
|
gmapi.subscribe(symbols=codes_str, frequency='60s', count=1,
|
unsubscribe_previous=True)
|
|
|
def recieve_msg(pipe):
|
while True:
|
value = pipe.recv()
|
print("跨进程通信：", value)
|
jsonValue = json.loads(value)
|
action = jsonValue["type"]
|
if action == 'resub':
|
re_subscribe_tick()
|
elif action == 'accpt_price':
|
try:
|
datas = jsonValue["data"]
|
for data in datas:
|
accpt_price(data["code"], float(data["price"]))
|
except Exception as e:
|
print(str(e))
|
|
|
class JueJinManager:
|
def __init__(self, pipe):
|
self.pipe = pipe
|
# 开启线程接受数据
|
t1 = threading.Thread(target=lambda: recieve_msg(pipe))
|
t1.setDaemon(True)
|
t1.start()
|
|
@classmethod
|
def get_gp_latest_info(cls, codes):
|
account_id, s_id, token = getAccountInfo()
|
symbols = gpcode_manager.get_gp_list_with_prefix(codes)
|
gmapi.set_token(token)
|
data = gmapi.get_instruments(symbols=",".join(symbols))
|
print(data)
|
return data
|
|
@classmethod
|
def get_now_price(cls, codes):
|
data = JueJinManager.get_gp_current_info(codes)
|
prices = []
|
for item in data:
|
code = item["symbol"].split(".")[1]
|
price = item["price"]
|
prices.append((code, price))
|
return prices
|
|
# 获取代码的涨幅
|
@classmethod
|
def get_codes_limit_rate(cls, codes):
|
datas = JueJinManager.get_gp_latest_info(codes)
|
pre_price_dict = {}
|
for data in datas:
|
code = data["sec_id"]
|
pre_close = tool.to_price(decimal.Decimal(str(data['pre_close'])))
|
pre_price_dict[code] = pre_close
|
|
now_prices = JueJinManager.get_now_price(codes)
|
f_results = []
|
for data in now_prices:
|
code = data[0]
|
price = data[1]
|
pre_price = float(pre_price_dict.get(code))
|
rate = round((price - pre_price) * 100 / pre_price, 2)
|
f_results.append((code, rate))
|
f_results.sort(key=lambda tup: tup[1])
|
f_results.reverse()
|
return f_results
|
|
@classmethod
|
def get_history_tick_n(cls, code, count, fields=None):
|
account_id, s_id, token = getAccountInfo()
|
symbols = gpcode_manager.get_gp_list_with_prefix([code])
|
gmapi.set_token(token)
|
results = gmapi.history_n(symbol=symbols[0], frequency="1d", count=count, fields=fields)
|
return results
|
|
@classmethod
|
def get_lowest_price_rate(cls, code, count):
|
datas = cls.get_history_tick_n(code, count)
|
low_price = datas[0]["close"]
|
for data in datas:
|
if low_price > data["close"]:
|
low_price = data["close"]
|
return (datas[-1]["close"] - low_price) / low_price
|
|
@classmethod
|
def get_gp_current_info(cls, codes):
|
account_id, s_id, token = getAccountInfo()
|
symbols = gpcode_manager.get_gp_list_with_prefix(codes)
|
gmapi.set_token(token)
|
data = gmapi.current(symbols=",".join(symbols))
|
print(data)
|
return data
|
|
@classmethod
|
def get_gp_codes_names(cls, codes):
|
datas = cls.get_gp_latest_info(codes)
|
results = {}
|
for data in datas:
|
code = data["symbol"].split(".")[1]
|
code_name = data['sec_name']
|
results[code] = code_name
|
return results
|
|
def start(self):
|
account_id, s_id, token = getAccountInfo()
|
gmapi.run(strategy_id=s_id,
|
filename='juejin.py',
|
# todo 回测模式
|
mode=gmapi.MODE_LIVE,
|
backtest_start_time='2022-08-18 09:25:00',
|
backtest_end_time='2022-08-18 10:30:00',
|
token=token)
|
|
def stop(self):
|
gmapi.stop()
|
|
@classmethod
|
def add_listen_code(cls, code):
|
redis = redisManager.getRedis()
|
redis.setex("juejin_listen_code-{}".format(code), 20, "1")
|
|
@classmethod
|
def get_listen_codes_lenth(cls):
|
redis = redisManager.getRedis()
|
keys = redis.keys("juejin_listen_code-*")
|
return len(keys)
|
|
@classmethod
|
def get_previous_trading_date(cls, date):
|
account_id, s_id, token = getAccountInfo()
|
gmapi.set_token(token)
|
return gmapi.get_previous_trading_date("SHSE", date)
|
|
|
def trade(code, volume):
|
account_id, s_id, token = getAccountInfo()
|
gmapi.set_token(token)
|
gmapi.set_account_id(account_id)
|
result = gmapi.order_volume(symbol=code, volume=volume, side=gmapi.OrderSide_Sell,
|
order_type=gmapi.OrderType_Market, position_effect=gmapi.PositionEffect_Close)
|
print(result)
|
|
|
# 获取近90天的最大量与最近的量
|
def get_volumns(codes):
|
end = datetime.datetime.now()
|
# 获取近90天的历史数据
|
cha = datetime.timedelta(days=90)
|
start = end - cha
|
|
account_id, s_id, token = getAccountInfo()
|
gmapi.set_token(token)
|
gmapi.set_account_id(account_id)
|
results = gmapi.history(symbol=gpcode_manager.get_gp_list_with_prefix(codes), frequency="1d",
|
start_time="{:%Y-%m-%d}".format(start),
|
fields="symbol,volume,eob",
|
end_time="{:%Y-%m-%d}".format(end))
|
_fresult = {}
|
|
for result in results:
|
code = result["symbol"].split(".")[1]
|
volumn = int(result["volume"])
|
day = "{:%Y-%m-%d}".format(result["eob"])
|
|
if _fresult.get(code) is None:
|
_fresult[code] = (volumn, volumn)
|
|
if volumn > _fresult[code][0]:
|
_fresult[code] = (volumn, _fresult[code][1])
|
_fresult[code] = (_fresult[code][0], volumn)
|
return _fresult
|
|
|
# 获取近90天的最大量与最近的量
|
# 获取最近一次涨停/涨停下一个交易日的最大值
|
def get_volumns_by_code(code, count=60) -> object:
|
datas = JueJinManager.get_history_tick_n(code, count, "open,high,low,close,volume,pre_close,bob")
|
# 计算
|
datas.sort(key=lambda x: x["bob"], reverse=True)
|
return datas
|
|
|
# 解析最大量
|
def parse_max_volume(datas, is_new_top=False):
|
max_volume = 0
|
|
max_volume_date = None
|
if is_new_top:
|
# 如果是突破前高就取最大量
|
for item in datas:
|
if max_volume < item["volume"]:
|
max_volume = item["volume"]
|
max_volume_date = item["bob"]
|
else:
|
date = None
|
for i in range(len(datas)):
|
# 查询涨停
|
item = datas[i]
|
volume = item["volume"]
|
if max_volume < volume:
|
max_volume = volume
|
max_volume_date = item['bob']
|
# 是否有涨停
|
limit_up_price = float(gpcode_manager.get_limit_up_price_by_preprice(item["pre_close"]))
|
if abs(limit_up_price - item["high"]) < 0.01:
|
next_volume = 0
|
if i > 0:
|
next_volume = datas[i - 1]["volume"]
|
date = datas[i]["bob"]
|
if volume < next_volume:
|
volume = next_volume
|
date = datas[i - 1]["bob"]
|
return volume, volume, date.strftime("%Y-%m-%d")
|
return max_volume, max_volume, max_volume_date.strftime("%Y-%m-%d")
|
|
|
# 是否有涨停
|
def is_have_limit_up(datas):
|
for i in range(len(datas)):
|
item = datas[i]
|
limit_up_price = float(gpcode_manager.get_limit_up_price_by_preprice(item["pre_close"]))
|
if abs(limit_up_price - item["close"]) < 0.01:
|
return True
|
return False
|
|
|
# 首板涨停溢价率
|
def get_limit_up_money_percent(datas):
|
datas.sort(key=lambda x: x["bob"])
|
first_rate_list = []
|
for i in range(0, len(datas)):
|
item = datas[i]
|
limit_up_price = float(gpcode_manager.get_limit_up_price_by_preprice(item["pre_close"]))
|
if abs(limit_up_price - item["close"]) < 0.001 and abs(
|
limit_up_price - datas[i - 1]["close"]) >= 0.001 and 0 < i < len(datas) - 1:
|
# 首板涨停
|
rate = (datas[i + 1]["high"] - datas[i + 1]["pre_close"]) / datas[i + 1]["pre_close"]
|
first_rate_list.append(rate)
|
if not first_rate_list:
|
return 1
|
count = 0
|
for rate in first_rate_list:
|
if rate >= 0.01:
|
count += 1
|
return count / len(first_rate_list)
|
|
|
# 根据涨幅高低分配交易窗口
|
def distribute_buy_win():
|
if tool.trade_time_sub(tool.get_now_time_str(), "09:30:00") > 0:
|
raise Exception("只能9:30之前重新分配窗口")
|
|
datas = JueJinManager.get_codes_limit_rate(gpcode_manager.get_gp_list())
|
matrix = numpy.array(datas)
|
codes = matrix[:, 0].tolist()
|
trade_gui.THSBuyWinManagerNew.fill_codes(codes)
|
|
|
if __name__ == '__main__':
|
datas = (get_volumns_by_code("603083", 150))
|
print(datas)
|
print(get_limit_up_money_percent(datas))
|
