ENE指标，自己百度
适用场景，在第二阶段上升趋势中，定点狙击ene底部的票，可以获得不错的收益
import schedule
import talib
import datetime
import time
import easyutils as eu
import tushare as ts
import numpy as np
import pandas as pd
import smtplib
from email.mime.text import MIMEText
from email.utils import formataddr
from retrying import retry
# ENE list
enelist_code = pd.Series(np.array([]))
enelist_upper = pd.Series(np.array([]))
enelist_lower = pd.Series(np.array([]))
enelist_ene = pd.Series(np.array([]))
enelistdata = pd.DataFrame({'code': enelist_code, 'upper': enelist_upper,
'ene': enelist_ene, 'lower': enelist_lower}, index=enelist_code)
sendlist = []
def getsma(stockdata, timeperiod=5):
ma = talib.SMA(stockdata['close'].values, timeperiod)
return ma.round(2)
@retry
def getstockdataofday(stockcode, timeperiod=180):
datedelay = datetime.date.today() - datetime.timedelta(days=timeperiod)
stockdata = ts.get_k_data(stockcode, start=str(datedelay.isoformat()),
end=str(datetime.date.today().isoformat()))
return stockdata
def getenevalues(stockdata):
if stockdata.empty:
return (-1, -1, -1)
param_m1 = 11
param_m2 = 9
param_n = 10
if stockdata is None:
return (-1, -1, -1)
if len(stockdata['close'].values) == 0:
return (-1, -1, -1)
sma_n = talib.SMA(stockdata['close'].values, param_n)
upper = (1 + param_m1 / 100) * sma_n
lower = (1 - param_m2 / 100) * sma_n
ene = (upper + lower) / 2
upper = upper.round(2)
ene = ene.round(2)
lower = lower.round(2)
stockdata['ene_upper'] = upper
stockdata['ene_ene'] = ene
stockdata['ene_lower'] = lower
upper_line = upper[-1]
lower_line = lower[-1]
ene_line = ene[-1]
return (upper_line, ene_line, lower_line)
@retry
def getrealquote(stockcode):
df = ts.get_realtime_quotes(stockcode)
return df
@retry
def mail(msgcontent):
ret = True
try:
msg = MIMEText(msgcontent, 'plain', 'utf-8')
# 括号里的对应发件人邮箱昵称、发件人邮箱账号
msg['From'] = formataddr(["From平凡啊菜", my_sender])
# 括号里的对应收件人邮箱昵称、收件人邮箱账号
msg['To'] = formataddr(["平凡啊菜", my_user])
msg['Subject'] = "ene指标监控，达到下轨的个股" # 邮件的主题，也可以说是标题
server = 465) # 发件人邮箱中的SMTP服务器，端口是25
server.login(my_sender, my_pass) # 括号中对应的是发件人邮箱账号、邮箱密码
# 括号中对应的是发件人邮箱账号、收件人邮箱账号、发送邮件
server.sendmail(
my_sender, [my_user, my_user1, my_user2], msg.as_string())
server.quit()
except Exception:
ret = False
return ret
def getready():
sendlist.clear()
i = 0
total = len(basic)
for stockcode in basic.index.values:
upper_line, ene_line, lower_line = getenevalues(
getstockdataofday(stockcode, 180))
enelistdata.loc[i] = {
'code': stockcode, 'upper': upper_line, 'ene': ene_line, 'lower': lower_line}
i = i + 1
print('%s/%s' % (i, total))
now = datetime.datetime.now()
enelistdata.to_pickle('enedatalist_%s_%s_%s.pick' %
(now.year, now.month, now.day))
def job():
is_tradetime = eu.is_tradetime_now()
is_holiday = eu.is_holiday_today()
# now = datetime.datetime.now()
# enelistdata = pd.read_pickle('enedatalist_%s_%s_%s.pick' %
# (now.year, now.month, now.day))
# enelistdata.index = enelistdata['code']
# print(enelistdata)
menelistdata = enelistdata.fillna(-1)
menelistdata.index = menelistdata['code']
# print(menelistdata)
if (not is_holiday) and (is_tradetime):
msgcontent = ''
count = len(basic.index.values)
mix = 0
for stockcode in basic.index.values:
rdata = getrealquote(stockcode)
if rdata is None:
continue
mdata = rdata.fillna(0)
trade = float(mdata['price'][0])
name = mdata['name'][0]
upper_line = float(menelistdata.ix[stockcode].ix['upper'])
ene_line = float(menelistdata.ix[stockcode].ix['ene'])
lower_line = float(menelistdata.ix[stockcode].ix['lower'])
print(upper_line, ene_line, lower_line)
now = mdata['time'][0]
issend = False
for item in sendlist:
if item == stockcode:
issend = True
if (trade <= lower_line) and (trade != 0) and (not issend):
msgcontent = msgcontent + '\n个股: %s,代码: %s,ENE下轨值: %s,当前价: %s,时间: %s\n' % (
name, stockcode, lower_line, trade, now)
msgcontent = msgcontent + '基本面数据：\n行业:%s,地区:%s,市盈率:%s,市净率:%s,流通股本:%s亿,总股本:%s亿,总资产:%s万,流动资产:%s万,固定资产:%s万,公积金:%s,每股公积金:%s万,每股收益:%s,每股净资产:%s,未分利润:%s万,每股未分配:%s,收入同比:%s%%,利润同比:%s%%,毛利率:%s%%,净利润率:%s%%,股东人数:%s,上市时间:%s\n' % (
basic.ix[stockcode].ix['industry'], basic.ix[stockcode].ix['area'], basic.ix[stockcode].ix['pe'], basic.ix[stockcode].ix['pb'], basic.ix[stockcode].ix[
'outstanding'], basic.ix[stockcode].ix['totals'], basic.ix[stockcode].ix['totalAssets'], basic.ix[stockcode].ix['liquidAssets'], basic.ix[stockcode].ix['fixedAssets'],
basic.ix[stockcode].ix['reserved'], basic.ix[stockcode].ix['reservedPerShare'], basic.ix[stockcode].ix['esp'], basic.ix[stockcode].ix['bvps'], basic.ix[stockcode].ix['undp'], basic.ix[stockcode].ix['perundp'], basic.ix[stockcode].ix['rev'], basic.ix[stockcode].ix['profit'], basic.ix[stockcode].ix['gpr'], basic.ix[stockcode].ix['npr'], basic.ix[stockcode].ix['holders'], basic.ix[stockcode].ix['timeToMarket'])
# print(msgcontent)
print('found one.')
sendlist.append(stockcode)
print('%s/%s' % (mix, count))
mix = mix + 1
if msgcontent != '':
ret = mail(msgcontent)
if ret:
print('邮件发送成功')
else:
print('邮件发送失败')
def main():
is_holiday = eu.is_holiday_today()
getready()
if not is_holiday:
schedule.every().day.at('8:00').do(getready)
schedule.every(5).seconds.do(job)
while True:
schedule.run_pending()
time.sleep(2)
if __name__ == '__main__':
my_sender = #发件人邮箱
my_pass = '' #QQ邮箱密钥
my_user =
# my_user1 =
my_user2 =
basic = ts.get_stock_basics()
classified = ts.get_st_classified()
main()