python制作一个股票盯盘系统

没有多少时间，这里简单写一下过程，主要是代码。

一、建立数据库，我采用sqlite3.

一个是记录股票的表，一个是设置股票股价预警的表。

二、采集股票股价变化到数据库，我这里用的是周一到周五开市时间，如果遇到其他时间就没有办法采集。

# -*- coding: utf-8 -*-
"""
Created on Tue Aug 19 16:38:35 2025@author: Yang
"""
import time
import efinance as ef
import random
from apscheduler.schedulers.blocking import BlockingScheduler
from wxauto import WeChat
import sqlite3
from pathlib import Pathwx = WeChat()
who = '楊斐'
db_filepath = Path(__file__).joinpath("../gp.db").resolve()def insertdb(gpdh,gpmc,zxj,gxsj,ts):conn = sqlite3.connect(db_filepath, timeout=10, check_same_thread=False)c = conn.cursor()insert_query = "INSERT INTO gp(gpdh,gpmc,zxj,gxsj,ts) VALUES(?,?,?,?,?);"insert_data = (gpdh,gpmc,zxj,gxsj,ts) #ts 买入为1，卖出为2c.execute(insert_query,insert_data)conn.commit()c.close()conn.closedef job():global whoconn = sqlite3.connect(db_filepath, timeout=10, check_same_thread=False)c = conn.cursor()sqlget = "select gpdh,gmjg,mcjg,cy from mai where cy <> 2 and gmjg > 0;" #cy为1持有看，2删除，0没有设置c.execute(sqlget)cursor = c.execute(sqlget)rows = cursor.fetchall()for gp in rows:quote = ef.stock.get_latest_quote(gp[0])time.sleep(random.uniform(1.5, 3.5))gpdh = quote.iloc[0, 0]gpmc = quote.iloc[0, 1]gpj = quote.iloc[0, 3]zxj = gpjgxsj = quote.iloc[0, 18]        gpj = quote[['最新价']].iloc[0, 0]if gpj <= gp[1]:print(gp[0]+'--' + str(gpj) + '元，可以买')message = f'{gp[0]}：最新股价{gpj}元，可以买了。原设置买入股价为{gp[1]}'wx.SendMsg(message, who)ts = 1elif gpj >= gp[2] and gp[3] == 1:print(gp[0]+'--' + str(gpj) + '元，可以卖出')message = f'{gp[0]}：最新股价{gpj}元，可以卖出。原设置卖出股价为{gp[2]}'wx.SendMsg(message, who)ts = 2else:print(gp[0]+'--' + str(gpj) + '元')ts = 0insertdb(gpdh,gpmc,zxj,gxsj,ts)job()scheduler = BlockingScheduler()# 上午时段(9:30-11:30)每隔10分钟执行
scheduler.add_job(job,'cron',day_of_week='mon-fri',hour='9',minute='31,36,41,46,51,56',
)
scheduler.add_job(job,'cron',day_of_week='mon-fri',hour='10',minute='1,6,11,16,21,26,31,36,41,46,51',
)
scheduler.add_job(job,'cron',day_of_week='mon-fri',hour='11',minute='1,6,11,16,21,26,31,59',
)
# 下午时段(13:00-15:00)每隔10分钟执行
scheduler.add_job(job,'cron',day_of_week='mon-fri',hour='13-14',minute='1,6,11,16,21,26,31,36,41,46,51,56',
)
scheduler.add_job(job,'cron',day_of_week='mon-fri',hour='15',minute='1',
)print("定时任务已启动...")
scheduler.start()job()

上面程序，在开市时间，每5分钟采集一次股票股价，到达设置买入价格或卖出价格就发送微信给我提醒我操作。

三、网页界面

# -*- coding: utf-8 -*-
"""
Created on Mon Aug 25 09:56:58 2025@author: Yang
"""from flask import Flask, render_template, request, redirect, url_for
from flask_sqlalchemy import SQLAlchemy
import reapp = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///C:/Users/Yang/.spyder-py3/gp.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)class gp(db.Model):__tablename__ = 'gp'id = db.Column(db.Integer, primary_key=True)gpdh = db.Column(db.String(10))gpmc = db.Column(db.String(20))zxj = db.Column(db.Numeric(precision=10, scale=2))gxsj = db.Column(db.String(30))ts = db.Column(db.Integer)class mai(db.Model):__tablename__ = 'mai'id = db.Column(db.Integer, primary_key=True)gpdh = db.Column(db.String(10))gmjg = db.Column(db.Numeric(precision=10, scale=2))mcjg = db.Column(db.Numeric(precision=10, scale=2))cy = db.Column(db.Integer, default=0)def validate_six_digit(value):return bool(re.fullmatch(r'^\d{6}$', value))def validate_price(value):try:float_value = float(value)return len(str(float_value).split('.')[1]) <= 2except ValueError:return False@app.route('/')
def index():page = request.args.get('page', 1, type=int)  # 获取页码参数，默认为1per_page = request.args.get('per_page', 20, type=int)  # 获取每页数量，默认为10products = gp.query.order_by(gp.id.desc()).paginate(page=page, per_page=per_page, error_out=False)return render_template('index.html', data=products)@app.route('/sz', methods=['GET', 'POST'])
def sz():if request.method == 'POST':six_digit = request.form.get('six_digit')price1 = request.form.get('price1')price2 = request.form.get('price2')is_checked = 1 if request.form.get('is_checked') == 'on' else 0  # 转换复选框值为1/0errors = []if not validate_six_digit(six_digit):errors.append("6位数字验证失败")if not validate_price(price1):errors.append("价格1格式错误")if not validate_price(price2):errors.append("价格2格式错误")if not errors:new_data = mai(gpdh=six_digit,gmjg=float(price1),mcjg=float(price2),cy=is_checked  # 使用转换后的值)#在添加之前将原有设置为2mai.query.filter_by(gpdh=f'{six_digit}').update({'cy': 2})db.session.commit()db.session.add(new_data)db.session.commit()return redirect(url_for('success'))else:products0 = mai.query.filter(mai.cy < 2).all()return render_template('sz.html', errors=errors, datagp=products0)products0 = mai.query.filter(mai.cy < 2).all()return render_template('sz.html', datagp=products0)@app.route('/success')
def success():return "数据提交成功！<a href='../sz'>返回</a>"if __name__ == '__main__':app.run(host='127.0.0.1',debug=True)

其中的index.html

<!DOCTYPE html>
<html>
<head><title>情况列表</title><link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
</head>
<body><div class="container mt-4"><h2>情况列表 <a href='./sz'>设置</a></h2><table class="table table-striped"><thead><tr><th>代号</th><th>名称</th><th>价格</th><th>更新时间</th><th>提示</th></tr></thead><tbody>{% for item in data.items %}<tr><td>{{ item.gpdh }}</td><td>{{ item.gpmc }}</td><td>{{ item.zxj }}</td><td>{{ item.gxsj }}</td><td>{% if item.ts == 0 %}--{% elif item.ts == 1 %}买入{% else %}卖出{% endif %}</td></tr>{% endfor %}</tbody></table><nav><ul class="pagination">{% if data.has_prev %}<li class="page-item"><a class="page-link" href="?page={{ data.prev_num }}">上一页</a></li>{% endif %}{% for page_num in data.iter_pages() %}{% if page_num %}<li class="page-item {% if page_num == data.page %}active{% endif %}"><a class="page-link" href="?page={{ page_num }}">{{ page_num }}</a></li>{% else %}<li class="page-item disabled"><span class="page-link">...</span></li>{% endif %}{% endfor %}{% if data.has_next %}<li class="page-item"><a class="page-link" href="?page={{ data.next_num }}">下一页</a></li>{% endif %}</ul></nav></div>
</body>
</html>

其中的sz.html代码

<!DOCTYPE html>
<html>
<head><title>设置</title><style>.error { color: red; }form { max-width: 500px; margin: 20px auto; }label { display: block; margin-top: 10px; }input { width: 100%; padding: 8px; }button { margin-top: 20px; padding: 10px 20px; }</style><link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
</head>
<body><div class="container mt-4"><h2>gp设置 <a href='../'>主页</a></h2><table class="table table-striped"><thead><tr><th>代号</th><th>购买价格</th><th>卖出价格</th><th>持有</th></tr></thead><tbody>{% for item in datagp %}<tr><td>{{ item.gpdh }}</td><td>{{ item.gmjg }}</td><td>{{ item.mcjg }}</td><td>{% if item.cy == 0 %}无{% elif item.cy == 1 %}有{% else %}无{% endif %}</td></tr>{% endfor %}</tbody></table>{% if errors %}<div class="error">{% for error in errors %}<p>{{ error }}</p>{% endfor %}</div>{% endif %}<form method="POST"><div><label>代号6位纯数字:</label><input type="text" name="six_digit" required></div><div><label>购买价格(带2位小数):</label><input type="number" step="0.01" name="price1" required></div><div><label>卖出价格(带2位小数):</label><input type="number" step="0.01" name="price2" required></div><div><label><input type="checkbox" name="is_checked">是否持有(1=是/0=否)</label></div><button type="submit">提交</button></form></div>
</body>
</html>

四、采用cpolar挂到公网，制作安卓app获取并显示网页

参考通过beeware制作安卓apk用于获取cpolar网址-CSDN博客