work_flow.py

# -*- encoding: UTF-8 -*-

import data_fetcher
import utils
import strategy.enter as enter
from strategy import turtle_trade
from strategy import backtrace_ma250
from strategy import breakthrough_platform
from strategy import parking_apron
from strategy import low_backtrace_increase
from strategy import keep_increasing
import tushare as ts
import push
import logging
import db
import time
import datetime
import urllib
import settings
import pandas as pd
import mail_util as mailUtil
import json
import re
import os
import Constants


def process():
    logging.info("************************ process start ***************************************")
    try:
        all_data = ts.get_today_all()
        subset = all_data[['code', 'name', 'nmc']]
        subset.to_csv(settings.config['stocks_file'], index=None, header=True)
        stocks = [tuple(x) for x in subset.values]
        statistics(all_data, stocks)
    except urllib.error.URLError as e:
        subset = pd.read_csv(settings.config['stocks_file'])
        subset['code'] = subset['code'].astype(str)
        stocks = [tuple(x) for x in subset.values]

    if utils.need_update_data():
        utils.prepare()
        data_fetcher.run(stocks)
        check_exit()

    strategies = {
        '海龟交易法则': turtle_trade.check_enter,
        '放量上涨': enter.check_volume,
        '突破平台': breakthrough_platform.check,
        '均线多头': keep_increasing.check,
        '无大幅回撤': low_backtrace_increase.check,
        '停机坪': parking_apron.check,
        '回踩年线': backtrace_ma250.check,
    }

    if datetime.datetime.now().weekday() == 0:
        strategies['均线多头'] = keep_increasing.check

    for strategy, strategy_func in strategies.items():
        check(stocks, strategy, strategy_func)
        time.sleep(2)
    

    if settings.config['save']['enable']:
        fileNames = list(filter(fi, os.listdir()))
        
        today = time.strftime("%Y-%m-%d", time.localtime()) 
        sampleFileName = f'{Constants.SAMPLE_PREFIX}:{today}.csv'
        toEmail(contentFile = sampleFileName, fileNames=fileNames)

        # 邮件已发， 文件删除
        # sample 文件
        if os.path.exists(sampleFileName):
            os.remove(sampleFileName)
        for fileName in fileNames:
            os.remove(fileName)
            logging.info('del {} succ'.format(fileName))

    logging.info("************************ process   end ***************************************")


def check(stocks, strategy, strategy_func):
    end = None
    m_filter = check_enter(end_date=end, strategy_fun=strategy_func)
    results = list(filter(m_filter, stocks))
    push.strategy('**************"{0}"**************\n{1}\n**************"{0}"**************\n'.format(strategy, results))
    push.saveStrategyRes(strategy = strategy,res = results)


def fi(name):
    today = time.strftime("%Y-%m-%d", time.localtime())
    return os.path.isfile(name) & (re.match(f'.*{today}.*', name) is not None) & (name.find(Constants.SAMPLE_PREFIX) == -1)


def toEmail(contentFile, fileNames):
    today = time.strftime("%Y-%m-%d", time.localtime()) 
    try:
        with open(contentFile,'r', encoding='utf-8') as file:
            message = file.read()
            mailUtil.send_email(from_user=settings.config['mail']['EMAIL_USER'], to_users=settings.config['mail']['EMAIL_TO'], subject=f'{today}_详情',content=message, filenames=fileNames)
    except FileNotFoundError:
        print('无法打开指定的文件!')
    except LookupError:
        print('指定了未知的编码!')
    except UnicodeDecodeError:
        print('读取文件时解码错误!')




def check_enter(end_date=None, strategy_fun=enter.check_volume):
    def end_date_filter(code_name):
        data = utils.read_data(code_name)
        if data is None:
            return False
        else:
            return strategy_fun(code_name, data, end_date=end_date)
        # if result:
        #     message = turtle_trade.calculate(code_name, data)
        #     push.strategy("{0} {1}".format(code_name, message))

    return end_date_filter


# 统计数据
def statistics(all_data, stocks):
    limitup = len(all_data.loc[(all_data['changepercent'] >= 9.5)])
    limitdown = len(all_data.loc[(all_data['changepercent'] <= -9.5)])

    up5 = len(all_data.loc[(all_data['changepercent'] >= 5)])
    down5 = len(all_data.loc[(all_data['changepercent'] <= -5)])

    def ma250(stock):
        stock_data = utils.read_data(stock)
        return enter.check_ma(stock, stock_data)

    ma250_count = len(list(filter(ma250, stocks)))

    msg = "涨停数：{}   跌停数：{}\n涨幅大于5%数：{}  跌幅大于5%数：{}\n年线以上个股数量：    {}"\
        .format(limitup, limitdown, up5, down5, ma250_count)
    push.statistics(msg)
    push.saveSampleRes(msg)


def check_exit():
    t_shelve = db.ShelvePersistence()
    file = t_shelve.open()
    for key in file:
        code_name = file[key]['code_name']
        data = utils.read_data(code_name)
        if turtle_trade.check_exit(code_name, data):
            push.strategy("{0} 达到退出条件".format(code_name))
            del file[key]
        elif turtle_trade.check_stop(code_name, data, file[key]):
            push.strategy("{0} 达到止损条件".format(code_name))
            del file[key]

    file.close()