Stock Monitoring and Notification Project

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

import datetime
import json
import requests
import pandas as pd

# 腾讯日线
def get_price_day_tx(code, end_date='', count=10, frequency='1d'):  # 日线获取
    unit = 'week' if frequency in '1w' else 'month' if frequency in '1M' else 'day'  # 判断日线，周线，月线
    if end_date:  end_date = end_date.strftime('%Y-%m-%d') if isinstance(end_date, datetime.date) else \
    end_date.split(' ')[0]
    end_date = '' if end_date == datetime.datetime.now().strftime('%Y-%m-%d') else end_date  # 如果日期今天就变成空
    URL = f'http://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={code},{unit},,{end_date},{count},qfq'
    st = json.loads(requests.get(URL).content);
    ms = 'qfq' + unit;
    stk = st['data'][code]
    buf = stk[ms] if ms in stk else stk[unit]  # 指数返回不是qfqday,是day
    df = pd.DataFrame(buf, columns=['time', 'open', 'close', 'high', 'low', 'volume'], dtype='float')
    df.time = pd.to_datetime(df.time);
    df.set_index(['time'], inplace=True);
    df.index.name = ''  # 处理索引
    return df


# 腾讯分钟线
def get_price_min_tx(code, end_date=None, count=10, frequency='1d'):  # 分钟线获取
    ts = int(frequency[:-1]) if frequency[:-1].isdigit() else 1  # 解析K线周期数
    if end_date: end_date = end_date.strftime('%Y-%m-%d') if isinstance(end_date, datetime.date) else \
    end_date.split(' ')[0]
    URL = f'http://ifzq.gtimg.cn/appstock/app/kline/mkline?param={code},m{ts},,{count}'
    st = json.loads(requests.get(URL).content);
    buf = st['data'][code]['m' + str(ts)]
    df = pd.DataFrame(buf, columns=['time', 'open', 'close', 'high', 'low', 'volume', 'n1', 'n2'])
    df = df[['time', 'open', 'close', 'high', 'low', 'volume']]
    df[['open', 'close', 'high', 'low', 'volume']] = df[['open', 'close', 'high', 'low', 'volume']].astype('float')
    df.time = pd.to_datetime(df.time);
    df.set_index(['time'], inplace=True);
    df.index.name = ''  # 处理索引
    df['close'][-1] = float(st['data'][code]['qt'][code][3])  # 最新基金数据是3位的
    return df


# sina新浪全周期获取函数，分钟线 5m,15m,30m,60m  日线1d=240m   周线1w=1200m  1月=7200m
def get_price_sina(code, end_date='', count=10, frequency='60m'):  # 新浪全周期获取函数
    frequency = frequency.replace('1d', '240m').replace('1w', '1200m').replace('1M', '7200m');
    mcount = count
    ts = int(frequency[:-1]) if frequency[:-1].isdigit() else 1  # 解析K线周期数
    if (end_date != '') & (frequency in ['240m', '1200m', '7200m']):
        end_date = pd.to_datetime(end_date) if not isinstance(end_date, datetime.date) else end_date  # 转换成datetime
        unit = 4 if frequency == '1200m' else 29 if frequency == '7200m' else 1  # 4,29多几个数据不影响速度
        count = count + (datetime.datetime.now() - end_date).days // unit  # 结束时间到今天有多少天自然日(肯定 >交易日)
        # print(code,end_date,count)
    URL = f'http://money.finance.sina.com.cn/quotes_service/api/json_v2.php/CN_MarketData.getKLineData?symbol={code}&scale={ts}&ma=5&datalen={count}'
    dstr = json.loads(requests.get(URL).content);
    # df=pd.DataFrame(dstr,columns=['day','open','high','low','close','volume'],dtype='float')
    df = pd.DataFrame(dstr, columns=['day', 'open', 'high', 'low', 'close', 'volume'])
    df['open'] = df['open'].astype(float);
    df['high'] = df['high'].astype(float);  # 转换数据类型
    df['low'] = df['low'].astype(float);
    df['close'] = df['close'].astype(float);
    df['volume'] = df['volume'].astype(float)
    df.day = pd.to_datetime(df.day);
    df.set_index(['day'], inplace=True);
    df.index.name = ''  # 处理索引
    if (end_date != '') & (frequency in ['240m', '1200m', '7200m']): return df[df.index <= end_date][
                                                                            -mcount:]  # 日线带结束时间先返回
    return df


def get_price(code, end_date='', count=10, frequency='1d', fields=[]):  # 对外暴露只有唯一函数，这样对用户才是最友好的
    xcode = code.replace('.XSHG', '').replace('.XSHE', '')  # 证券代码编码兼容处理
    xcode = 'sh' + xcode if ('XSHG' in code) else 'sz' + xcode if ('XSHE' in code) else code

    if frequency in ['1d', '1w', '1M']:  # 1d日线  1w周线  1M月线
        try:
            return get_price_sina(xcode, end_date=end_date, count=count, frequency=frequency)  # 主力
        except:
            return get_price_day_tx(xcode, end_date=end_date, count=count,
                                    frequency=frequency)  # 备用

    if frequency in ['1m', '5m', '15m', '30m', '60m']:  # 分钟线 ,1m只有腾讯接口  5分钟5m   60分钟60m
        if frequency in '1m': return get_price_min_tx(xcode, end_date=end_date, count=count, frequency=frequency)
        try:
            return get_price_sina(xcode, end_date=end_date, count=count, frequency=frequency)  # 主力
        except:
            return get_price_min_tx(xcode, end_date=end_date, count=count, frequency=frequency)  # 备用

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

from email.header import Header
from email.mime.text import MIMEText

from Ashare import *
import smtplib
import os


def send_email(body):
    smtp_mail = os.environ.get('SMTP_MAIL')
    smtp_token = os.environ.get('SMTP_TOKEN')

    # 设置发件人和收件人的邮箱地址
    sender = smtp_mail
    recipient = smtp_mail

    # 设置邮件主题
    subject = "Github Actions: stock-spy Repository Remind"

    msg = MIMEText(body, "plain", "utf-8")
    msg["Subject"] = Header(subject, "utf-8")
    msg["from"] = sender
    msg["to"] = recipient

    # 连接到SMTP服务器
    server = smtplib.SMTP("smtp.qq.com")

    # 登录到SMTP服务器
    status, response = server.login(smtp_mail, smtp_token)

    if status == 235:
        print("SMTP 登录成功")
        if body != '':
            # 发送邮件
            server.sendmail(sender, recipient, msg.as_string())
            print("邮件已发送")
        else:
            print("无需发送邮件")
    else:
        print("SMTP 登录失败")

    # 关闭连接
    server.quit()


if __name__ == '__main__':
    dfTenYears = get_price('sh000001', frequency='1d', count=2500)

    lowLine = dfTenYears['close'].mean()
    highLine = lowLine + 470

    dfOneDay = get_price('sh000001', frequency='1d', count=1)

    lowPoint = dfOneDay['low'].mean()
    highPoint = dfOneDay['high'].mean()

    body = f"当前高水位是：{highLine}\n" \
           f"当前收盘价是：{dfOneDay['close'].mean()}\n" \
           f"当前低水位是：{lowLine}"

    print(body)

    if lowPoint < lowLine:
        send_email(f"当前最低点是：{lowPoint}\n"
                   f"当前低水位是：{lowLine}\n"
                   f"该买了")
    elif highPoint > highLine:
        send_email(f"当前最高点是：{highPoint}\n"
                   f"当前高水位是：{highLine}\n"
                   f"该卖了")
    else:
        send_email('')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

name: Python application

on:
  schedule:
    - cron: '0 8 * * *'
  workflow_dispatch:

permissions:
  contents: read

jobs:
  run_python_script:

    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v3
    - name: Set up Python 3.10
      uses: actions/setup-python@v3
      with:
        python-version: "3.10"
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install flake8 pytest
        python -m pip install requests pandas
        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi        
    - name: Set Environment Variables
      run: |
        echo "SMTP_MAIL=${{ secrets.SMTP_MAIL }}" >> $GITHUB_ENV
        echo "SMTP_TOKEN=${{ secrets.SMTP_TOKEN }}" >> $GITHUB_ENV        
    - name: Run Python script
      run: |
                python main.py