跳到主要内容

5 篇博文 含有标签「项目」

查看所有标签

  • 这是网盘爬取的部分记录，很多坑，分享一下避免大家采坑。
  • 后续发现网盘有网页版本，在安装能 root 的 Android 模拟器后，放弃通过 app 爬。

Fiddler​

教程

  • 电脑管家下的 Fiddler 4.0
    • Fiddler Anywhere 只找到苹果的破解版
  • 允许远程连接
    • 更改 fiddler 端口后要重启软件
  • 手机设置
    • 证书安装
    • 卸载方式
      • 搜索 加密与凭据
      • 用户凭据 找到自己命名的凭据 (fiddler) 点击删除

打开网盘后 无连接​

走 fiddler 后，app 打开显示无网络。用 adb 查看日志，找到如下错误：

06-06 19:09:03.624 28223 28223 W System.err: javax.net.ssl.SSLHandshakeException: java.security.cert.CertPathValidatorException: Trust anchor for certification path not found.

是因为，fiddler 装的证书不是根证书，app 不认可。 改为弄根证书需要手机 root，懒得弄了。

装安卓虚拟机​

考虑使用安卓模拟器

  • 夜神模拟器
    • hyper-v 冲突
  • bluestacks
  • 天天模拟器
    • 启动不了
    • 安装时蓝屏

卸载某个模拟器后 dll 乱了​

无效的 wechatwin.dll 文件 errcode:126, 点击“确定”下载最新版本

  • 微信打不开 怀疑是某个模拟器卸载的时候 把系统的 dll 一起
  • 修复方法参考

按 Win + S 键搜索栏输入 CMD，以管理员身份打开 “命令提示符” 输入命令：

sfc /SCANNOW

等待扫描结束：

Dism /Online /Cleanup-Image /ScanHealth

等进度条走完，再次打开微信，看是否能够正常打开。如果能正常打开，重启电脑再看是否能打开微信，如果重启之后还能正常打开微信等其它软件，说明此问题就修复了。

Hyper-V + Visual Studio 跑 Android 模拟器​

参考 用 Xamarin 的模拟器。

安装 Visual Studio 2022 一样可以

  • 打开 SDK 管理器后 提示：
    • Loading SDK component information failed. Please retry
    • 加载 SDK 组件信息失败 请重试
    • must not be an empty string forPath

通过点击：工具 - 选项 - 搜索 Android - Xamarin Android 设置 点击确定后就能找找到位置了。

下载 SDK 需要全局翻墙。

  • 爬完后注意存档位置 方便后续删除
    • C:\Users\xxx\AppData\Local\Xamarin\Mono for Android\Archives

如此，终于有一个能用的，和 hyper-v 不冲突的 Android 模拟器了。

到这里后，发现有网页版本，满足需求，放弃了 app 爬取。

询问大佬们的解决方案​

项目阅读需 3 分钟

lz18届毕业生一枚，工作有段时间了，计科专业。 上班的时候有很多想法，但没有时间实现。所以组个纯网络的兴趣小组，想把想法变现。

社团性质：

  • 纯网络兴趣小组（也就是没有工作室啥的，靠自己）
  • 纯公益社团（项目公益，不向社员收费同时也不赚钱，有能力可以靠自己接case）
  • python语言向（因为lz只会py编程语言，所以只能给这个支持）

小组成员事物内容：

  • 学习编程（偏python，因为lz只能给这方面的指导。。。）
  • 实现各种各样的idea（流程一般是：提出想法，大家讨论可行性，组队，实现想法）

小组需要：

  • 对编程有兴趣的人（估摸着应该靠自学，编程没兴趣学不下去的）
  • 有一定的经济基础（自己的电脑得跑得起编辑器把，租个服务器或买个树莓派啥的，但不会有小组收钱的行为）
  • 有时间（课程不紧，闲暇时间比较多）
  • 能自己写总结（分享型小组嘛）

加分项

  • 有一定编程基础
  • 计算机专业大二大三
  • 有一定的英语能力（编程，英语还是很重要的）

相信有些人觉得在学校里学的东西出来难找工作，我觉得做些开源项目会对以后找ITl类工作or实习有帮助，不过也不绝对，还是得靠自己努力。

Q群号： 870632533 欢迎转给身边人

项目阅读需 2 分钟

项目介绍​

通过Chrome与selenium 将网易云私信中歌曲放到一个歌单中，方便实时收听新歌。 我们可以直接用已登录的Chrome（保存有网易云网站登录态），模拟用户操作，收藏新歌。

image.png

关键代码​

  • 初始化selenium
def init_driver():    
    executable_path = "chromedriver"    
    if not os.path.exists("chromedriver.exe"):  # 无驱则需要下载    
        # 本地项目 复制驱动    
        _path = os.path.abspath(__file__)    
        if not "crawler_set" in _path:    
            root_path = _path[:_path.index("crawler_set") + len("crawler_set")]    
            executable_path = os.path.join(root_path, "driver", "chromedriver.exe")    
        else:  # 单个文件 下载驱动    
            url = "https://raw.githubusercontent.com/AngusWG/crawler_set/master/driver/chromedriver.exe"    
            proxies = {    
                "http": "socks5://127.0.0.1:1080",    
                "https": "socks5://127.0.0.1:1080"    
            }    
            r = requests.get(url, proxies=proxies)    
            with open("chromedriver.exe", "wb") as code:    
                content_size = int(r.headers['content-length'])  # 内容体总大小    
                for data in tqdm(iterable=r.iter_content(1024), total=content_size, unit="k", desc="下载驱动"):    
                    code.write(data)    
    
    user_cookies = "".join([os.path.expanduser('~'), r"\AppData\Local\Google\Chrome\User Data"])    
    
    option = webdriver.ChromeOptions()    
    option.add_argument("--user-data-dir={}".format(user_cookies))  # 设置成用户自己的数据目录    
    
    try:    
        driver = webdriver.Chrome(executable_path, options=option)    
        driver.implicitly_wait(5)    
        return driver    
    except WebDriverException:    
        print("请先关掉所有的Chrome")    
        exit(-2)
  • 收藏歌曲
def save_song(url):    
    print("[{}] start".format(url), end=" ")    
    driver.get(url)    
    driver.switch_to.frame("contentFrame")    
    title = driver.find_element_by_xpath('//div[contains(@class, "tit")]').text    
    for word in shielding_words:    
        if word in title:    
            print("{} 因 {} 已忽略".format(title, word))    
            return    
    driver.find_element_by_xpath('//*[contains(text(), "收藏")]').click()    
    driver.find_element_by_xpath('//*[contains(text(), "{}")]'.format(song_dir)).click()    
    print(title)

全部代码​

#!/usr/bin/python3    
# encoding: utf-8     
# @Time    : 2019/12/19 9:36    
# @author  : zza    
# @Email   : 740713651@qq.com    
# @File    : 将网易云的私信音乐整理.py    
"""    
该脚本会保存每天新私信里最后几首歌曲到tmp_save_dir    
1.在Chrome上登录自己的网易云帐号t    
2.创建tmp_save_dir    
然后运行脚本    
"""    
import os    
    
import requests    
from selenium import webdriver    
from selenium.common.exceptions import WebDriverException, NoSuchElementException    
from tqdm import tqdm    
    
song_dir = "tmp_save_dir"    
# 屏蔽关键词    
shielding_words = ['伴奏']    
    
    
def init_driver():    
    executable_path = "chromedriver"    
    if not os.path.exists("chromedriver.exe"):  # 无驱则需要下载    
        # 本地项目 复制驱动    
        _path = os.path.abspath(__file__)    
        if not "crawler_set" in _path:    
            root_path = _path[:_path.index("crawler_set") + len("crawler_set")]    
            executable_path = os.path.join(root_path, "driver", "chromedriver.exe")    
        else:  # 单个文件 下载驱动    
            url = "https://raw.githubusercontent.com/AngusWG/crawler_set/master/driver/chromedriver.exe"    
            proxies = {    
                "http": "socks5://127.0.0.1:1080",    
                "https": "socks5://127.0.0.1:1080"    
            }    
            r = requests.get(url, proxies=proxies)    
            with open("chromedriver.exe", "wb") as code:    
                content_size = int(r.headers['content-length'])  # 内容体总大小    
                for data in tqdm(iterable=r.iter_content(1024), total=content_size, unit="k", desc="下载驱动"):    
                    code.write(data)    
    
    user_cookies = "".join([os.path.expanduser('~'), r"\AppData\Local\Google\Chrome\User Data"])    
    
    option = webdriver.ChromeOptions()    
    option.add_argument("--user-data-dir={}".format(user_cookies))  # 设置成用户自己的数据目录    
    
    try:    
        driver = webdriver.Chrome(executable_path, options=option)    
        driver.implicitly_wait(5)    
        return driver    
    except WebDriverException:    
        print("请先关掉所有的Chrome")    
        exit(-2)    
    
    
def get_private_detail():    
    # 点击私信后    
    driver.get("https://music.163.com/#/msg/m/private")    
    driver.switch_to.frame("contentFrame")    
    new_msg_items = driver.find_elements_by_xpath('//i[@class="u-bub"]/b[@class="f-alpha"]/..//parent::*//a')    
    private_detail_url_dict = dict()    
    for i in new_msg_items:    
        _, singer_id = i.get_attribute("href").split("?")    
        uri = "https://music.163.com/#/msg/m/private_detail?" + singer_id    
        msg_num = int(i.find_element_by_xpath("../i/em").text)    
        private_detail_url_dict[uri] = msg_num    
    return private_detail_url_dict    
    
    
def get_song_url_from_album_set(url):    
    song_set = set()    
    # 歌曲页面保存    
    driver.get(url)    
    driver.switch_to.frame("contentFrame")    
    url_list = driver.find_elements_by_xpath('//a[contains(@href, "/song?id")]')    
    for item in url_list:    
        if "伴奏" in item.text:    
            break    
        _, song_id = item.get_attribute("href").split("?id=")    
        song_set.add("https://music.163.com/#/song?id=" + song_id)    
        song_name = item.find_element_by_xpath("./b").get_attribute("title")    
        print(song_name, end=" ")    
    return song_set    
    
    
def get_song_url_from_private_detail(url, msg_num):    
    album_set = set()    
    song_set = set()    
    # 歌曲页面保存    
    driver.get(url)    
    driver.switch_to.frame("contentFrame")    
    url_list = driver.find_elements_by_xpath('//div[contains(@class,"itemleft")]')[-msg_num:]    
    for item in url_list:    
        try:    
            i = item.find_element_by_xpath(    
                './/a[contains(@href,"album?id") or contains(@href, "song?id")]').get_attribute("href")    
            _, _id = i.split("?id=")    
            if "song?id" in i:    
                song_set.add("https://music.163.com/#/song?id=" + _id)    
            else:  # "album?id"    
                album_set.add("https://music.163.com/#/album?id=" + _id)    
        except NoSuchElementException:    
            pass    
    
    for album_url in album_set:    
        song_set.update(get_song_url_from_album_set(album_url))    
    return song_set    
    
    
def save_song(url):    
    print("[{}] start".format(url), end=" ")    
    driver.get(url)    
    driver.switch_to.frame("contentFrame")    
    title = driver.find_element_by_xpath('//div[contains(@class, "tit")]').text    
    for word in shielding_words:    
        if word in title:    
            print("{} 因 {} 已忽略".format(title, word))    
            return    
    driver.find_element_by_xpath('//*[contains(text(), "收藏")]').click()    
    driver.find_element_by_xpath('//*[contains(text(), "{}")]'.format(song_dir)).click()    
    print(title)    
    
    
driver = init_driver()    
if not os.path.exists("tmp.txt"):    
    # 获取私信用户列表    
    private_detail_url_dict = get_private_detail()    
    print("private_detail_url_set len={}".format(len(private_detail_url_dict)))    
    # 获取歌曲id    
    song_url_set = set()    
    for private_detail_url, msg_num in private_detail_url_dict.items():    
        song_url_set.update(get_song_url_from_private_detail(private_detail_url, msg_num))    
        print("song_url_set len={}".format(len(song_url_set)))    
    with open("tmp.txt", "w", encoding="utf8") as f:    
        f.write("\n".join(song_url_set))    
else:    
    with open("tmp.txt", "r", encoding="utf8") as f:    
        data = f.read()    
    song_url_set = data.split("\n") if data else []    
# 保存歌曲    
for song_url in song_url_set:    
    save_song(song_url)    
os.remove("tmp.txt")    
driver.close()

项目阅读需 3 分钟

写在前面的话​

  • 通过 米筐 网站提供的量化平台进行的测试。
  • 做这个研究的初衷是笔者有一点意象投资股市，有几个不同的投资方案（策略），我想先拿 19 年的数据跑一跑我的方案（策略），如果亏钱了，就不玩股市，老老实实玩基金了。

关于如何读这个图​

  • 首先能看到时间，我的数据是从2019 年 01 -- 2020 年 01,
  • 红色的线代表账户市值（把股票持仓也换算成钱）的变化。
  • 蓝色的线代表基准（沪深三百）的市值变化。
  • 其次最重要的指标，回测收益，表示的是我本金到时间结束，涨了多少，比如我有 5w，那么这个图表示我年末结束涨到了 5*(0.09+1) = 5.45（万元）。
  • 然后平均一年能有 13.5%的收益。
  • 然后看基准收益, 这个我一般设置的是沪深三百，也就是大盘，这个图表示 19 年大盘涨了30+%，而我的策略只有9+%，其实是没有跑赢大盘的 (还不如买基金)。
  • 然后需要注意下一个指标"MaxDrawdown（最大回测）": 最大回撤是最常用的指标，描述了投资者可能面临的最大亏损。最大回撤的数值越小越好，越大说明风险越大。
  • Sharpe（夏普率）：夏普比率若为正值，代表基金报酬率大于风险；若为负值，代表基金风险大于报酬率。因此，夏普比率越高，投资组合越佳。

好了，下面开始正文。

检测低于 x 元的 股票 ，购买并持有 3 个礼拜后抛出​

这个是一个普遍的购买思路，觉得底价股票一定会长，我短期持有然后等他高于某个价格就出售，我模拟一下这个情况，下面是代码：

#!/usr/bin/python3    
# encoding: utf-8     
# @Time    : 2020/11/18 14:32    
# @author  : zza    
# @Email   : 740713651@qq.com    
    
import warnings    
    
import pandas    
import rqdatac    
    
__config__ = {    
    "base": {    
        "start_date": "20190101",    
        "end_date": "20191231",    
        'frequency': '1d',    
        "accounts": {    
            "stock": 5000,    
        }    
    },    
    "mod": {    
        "sys_progress": {    
            "enabled": True,    
            "show": True    
        }, "sys_accounts": {    
            "enabled": True,    
            "dividend_reinvestment": True,    
        },    
        "sys_analyser": {    
            "enabled": True,    
            "plot": True,    
            'benchmark': '000300.XSHG',    
        },    
    }    
}    
    
def init(context):    
    context.lowest = 5    
    context.highest = 5.9    
    context.day_count = 0    
    context.holding_days = 10    
    
def get_ticker(context):    
    all_ins = rqdatac.all_instruments("Stock").order_book_id.to_list()    
    price_df = rqdatac.get_price(all_ins, context.now, context.now, expect_df=True, fields=['close'])    
    price_df = price_df.reset_index(drop=False)    
    price_df = price_df[(context.lowest <= price_df.close) & (price_df.close <= context.highest)]    
    
    start_date = rqdatac.get_previous_trading_date(context.now, 2)    
    price_df_2 = rqdatac.get_price(price_df.order_book_id.to_list(), start_date, context.now, expect_df=True,    
                                   fields=['close']).reset_index(drop=False)    
    order_ticker = []    
    for order_book_id, data_df in price_df_2.groupby(price_df_2.order_book_id):    
        if (data_df.close.shift(fill_value=0) < data_df.close).all():    
            order_ticker.append(order_book_id)    
    
    return order_ticker    
    
def before_trading(context):    
    context.day_count += 1    
    if (context.day_count - 1) % context.holding_days == 0:    
        context.order_ticker = get_ticker(context)    
    
def handle_bar(context, bar_dict):    
    if (context.day_count - 1) % context.holding_days == 0:    
        with warnings.catch_warnings():    
            for item in context.order_ticker:    
                o = order_target_value(item, 1000)    
                if o:    
                    print(o)    
    
    if (context.day_count - 1) % context.holding_days == (context.holding_days-1):    
        for item in context.portfolio.positions.keys():    
            order_target_value(item, 0)    
    
def after_trading(context):    
    print("total_value {}".format(context.portfolio.total_value))    
    if (context.day_count - 1) % 15 == 0:    
        items = []    
        for k, v in context.portfolio.positions.items():    
            items.append({"order_book_id": k, "quantity": v.quantity})    
        print(pandas.DataFrame(items))    
    
if __name__ == '__main__':    
    import rqalpha    
    
    rqalpha.run_func(init=init,    
                     before_trading=before_trading,    
                     handle_bar=handle_bar,    
                     after_trading=after_trading,    
                     config=__config__)

  • 持有 3 个交易日 image.png

  • 持有 10 个交易日

  • 持有 15 个交易日

结论：底价股票确实会随着大盘的涨而有一定的波动，但最终都是会赔钱的，且三个交易的日的时候，直接赔钱了。这个图可以给广大读者的保佑侥幸心里的朋友看看，一年忙到头，不如买国运（沪深三百）。

根据公众号 闷声发小财​

这个是根据我一个朋友跑的算法跑出来的可能会涨的概率。笔者拿到了他一年的数据（没用爬虫），然后试试能不能涨，这么跑的原因是，我当初确实想按着他的这个公众号来买，不过既然有回放数据测试，我当然先测试一下，本金本来就不多，挥霍不了，不敢随便教学费。

策略如下：

#!/usr/bin/python3    
# encoding: utf-8     
# @Time    : 2020/11/22 16:33    
# @author  : zza    
# @Email   : 740713651@qq.com    
# @File    : 闷声发小财。py    
    
import warnings    
    
import numpy    
import pandas    
import rqdatac    
from rqalpha.apis import *    
    
__config__ = {    
    "base": {    
        "start_date": "20190101",    
        "end_date": "20191231",    
        'frequency': '1d',    
        "accounts": {    
            "stock": 5000,    
        },    
        "data_bundle_path": r"E:\data\bundle",    
    },    
    "extra": {    
        "log_level": "debug",    
    },    
    "mod": {    
        "sys_progress": {    
            "enabled": True,    
            "show": True    
        }, "sys_accounts": {    
            "enabled": True,    
            "dividend_reinvestment": True,    
        },    
        "sys_analyser": {    
            "enabled": True,    
            "plot": True,    
            'benchmark': '000300.XSHG',    
        },    
    },    
}    
    
def init(context):    
    context.df = pandas.read_csv("发小财。csv", dtype={"股票代码": numpy.str})    
    context.df['order_book_id'] = rqdatac.id_convert(context.df["股票代码"].to_list())    
    context.used_df = context.df[context.df["预测结果"] > 0.7]    
    context.sell_multiple = 1.08    
    context.holding_days = 1    
    context.buy_queen = []    
    context.sell_map = {}    
    
def before_trading(context):    
    print("当日购买：{}".format(context.buy_queen))    
    print("当日持仓：{}".format(list(context.sell_map.keys())))    
    
def handle_bar(context, bar_dict):    
    while context.buy_queen:    
        order_book_id = context.buy_queen.pop()    
        if context.stock_account.cash > 1000:    
            o = order_value(order_book_id, 1000)    
            if o:    
                trading_dt = get_next_trading_date(context.now, context.holding_days).to_pydatetime()    
                context.sell_map[order_book_id] = trading_dt    
                print("[{}] 购买成功".format(order_book_id))    
        else:    
            print("[{}] 资金不足，无法购买".format(order_book_id))    
    for order_id, dt in context.sell_map.copy().items():    
        a = dt.date() <= context.now.date()    
        b = bar_dict[order_id].close > get_position(order_id).avg_price # * context.sell_multiple    
        if a and b:    
            o = order_percent(order_id, -1)    
            if o:    
                del context.sell_map[order_id]    
                print("[{}] 卖出成功".format(order_id))    
    
def dt_to_int(dt):    
    return dt.year * 10000 + dt.month * 100 + dt.day    
    
def after_trading(context):    
    df = context.used_df[dt_to_int(context.now) == context.used_df.trading_dt]    
    if df.empty:    
        return    
    else:    
        for _, item in df.iterrows():    
            context.buy_queen.append(item["order_book_id"])    
    
if __name__ == '__main__':    
    import rqalpha    
    
    rqalpha.run_func(init=init,    
                     before_trading=before_trading,    
                     handle_bar=handle_bar,    
                     after_trading=after_trading,    
                     config=__config__)

  • 持有一天就卖出 image.png

  • 持有 3 天并价格高于持仓价则卖出

image.png

结论：还好跑了回放测试，不然得教好多学费。

检测行业因子普遍 (90%) 一周内上涨的低价股票​

这个也是我在上班的时候偶尔能想到的，我觉得可能也代表一分部分人的想法，**某个行业市场涨幅的时候，必定是全行业性质的，且会持续一段时间。**于是我就写了这个，根据股票行业暴露度（属于哪个行业），分类行业，如果都涨了，就买进，如果都跌了，就卖出。

  • 检测行业因子普遍 (90%) 一周内上涨的低价股票 购买并持有
  • 在行业下跌在 30% 左右卖出

代码如下：

#!/usr/bin/python3    
# encoding: utf-8     
# @Time    : 2020/11/22 16:33    
# @author  : zza    
# @Email   : 740713651@qq.com    
# @File    : 闷声发小财。py    
from collections import defaultdict    
from pprint import pprint    
    
import numpy    
import pandas    
from rqalpha.apis import *    
    
__config__ = {    
    "base": {    
        "start_date": "20190101",    
        "end_date": "20191231",    
        'frequency': '1d',    
        "accounts": {    
            "stock": 50000,    
        },    
        "data_bundle_path": r"E:\data\bundle",    
    },    
    "extra": {    
        "log_level": "debug",    
    },    
    "mod": {    
        "sys_progress": {    
            "enabled": True,    
            "show": True    
        }, "sys_accounts": {    
            "enabled": True,    
            "dividend_reinvestment": True,    
        },    
        "sys_analyser": {    
            "enabled": True,    
            "plot": True,    
            'benchmark': '000300.XSHG',    
        },    
    },    
}    
    
def get_holding_ticker():    
    # 请去米筐购买或申请试用 RQData : www.ricequant.com    
    rqdatac.init()    
    df = rqdatac.get_factor_exposure(rqdatac.all_instruments("Stock").order_book_id, "20190101", "20191231")    
    df.reset_index().sort_values("date").to_csv("factor_exposure.csv", index=False)    
    
def init(context):    
    df = pandas.read_csv("factor_exposure.csv")    
    all_order_book_id = all_instruments().order_book_id.to_list()    
    context.df = df[df["order_book_id"].isin(all_order_book_id)]    
    context.buy_map = defaultdict(list)    
    context.holding_map = defaultdict(list)    
    context.sell_factors = []    
    context.day_count = 0    
    context.sell_multiple = 1.2    
    
def before_trading(context):    
    pprint("当日购买行业：{}".format(context.buy_map))    
    print("当日持仓：{}".format([i.order_book_id for i in get_positions()]))    
    print("年化：{}".format(context.portfolio.total_returns))    
    
def handle_bar(context, bar_dict):    
    # buy    
    for factors, order_book_ids in context.buy_map.items():    
        if context.stock_account.cash < 2000:    
            break    
        for order_book_id in order_book_ids.copy():    
            o = order_value(order_book_id, 2000)    
            if o:    
                context.holding_map[factors].append(order_book_id)    
                order_book_ids.remove(order_book_id)    
    # sell    
    for factors in context.sell_factors:    
        for order_book_id in context.holding_map[factors].copy():    
            # a =  bar_dict[order_book_id].close > get_position(order_book_id).avg_price * context.sell_multiple    
            a = True    
            if a:    
                o = order_percent(order_book_id, -1)    
                if o:    
                    context.holding_map[factors].remove(order_book_id)    
    
def dt_to_str(dt):    
    return "{}-{}-{}".format(dt.year, dt.month, dt.day)    
    
def is_rise(order_book_id):    
    a, b = history_bars(order_book_id, 2, frequency="1d", fields='close')    
    return b > a    
    
def after_trading(context):    
    context.day_count += 1    
    if context.day_count % 5 != 1:    
        return    
    df = context.df[context.now.date().isoformat() == context.df.date].copy()    
    df.loc[:, 'rose'] = df["order_book_id"].apply(is_rise)    
    df.loc[:, 'close'] = df["order_book_id"].apply(lambda x: history_bars(x, 1, frequency="1d", fields='close'))    
    context.sell_factors = []    
    context.buy_map = defaultdict(list)    
    for factors in df.columns[13:-2]:    
        _df = df[df[factors] == 1]    
        if _df['rose'].sum() / _df['order_book_id'].size > 0.9:    
            context.buy_map[factors] = _df[(_df["close"] > 5) & (_df["close"] < 6)].sort_values('close')[    
                                           'order_book_id'].to_list()[:2]    
        elif _df['rose'].sum() / _df['order_book_id'].size < 0.3:    
            context.sell_factors.append(factors)    
    del df    
    
if __name__ == '__main__':    
    import rqalpha    
    
    rqalpha.run_func(init=init,    
                     before_trading=before_trading,    
                     handle_bar=handle_bar,    
                     after_trading=after_trading,    
                     config=__config__)
  • 各行业两只股票

emmm，不说了，都是磊。

普通 macd 均线策略​

  • 这个是米筐官网的默认策略
from collections import defaultdict    
from pprint import pprint    
    
import numpy    
import pandas    
from rqalpha.apis import *    
    
__config__ = {    
    "base": {    
        "start_date": "20190101",    
        "end_date": "20191231",    
        'frequency': '1d',    
        "accounts": {    
            "stock": 50000,    
        },    
        "data_bundle_path": r"E:\data\bundle",    
    },    
    "extra": {    
        "log_level": "debug",    
    },    
    "mod": {    
        "sys_progress": {    
            "enabled": True,    
            "show": True    
        }, "sys_accounts": {    
            "enabled": True,    
            "dividend_reinvestment": True,    
        },    
        "sys_analyser": {    
            "enabled": True,    
            "plot": True,    
            'benchmark': '000300.XSHG',    
        },    
    },    
}    
    
import talib    
    
def init(context):    
    context.s1 = '000300.XSHG'    
    
    context.SHORTPERIOD = 12    
    context.LONGPERIOD = 26    
    context.SMOOTHPERIOD = 9    
    context.OBSERVATION = 100    
    
    subscribe(context.s1)    
    
def before_trading(context):    
    pass    
    
def handle_bar(context, bar_dict):    
    closes = history_bars(context.s1, context.OBSERVATION, '1d', 'close')    
    
    diff, signal, _ = talib.MACD(closes)    
    
    if diff[-1] > signal[-1] and diff[-2] < signal[-2]:    
        order_target_percent(context.s1, 1)    
    
    if diff[-1] < signal[-1] and diff[-2] > signal[-2]:    
        order_target_percent(context.s1, 0)    
    
def after_trading(context):    
    pass    
    
if __name__ == '__main__':    
    import rqalpha    
    
    rqalpha.run_func(init=init,    
                     before_trading=before_trading,    
                     handle_bar=handle_bar,    
                     after_trading=after_trading,    
                     config=__config__)

image.png

这个是根据大佬的代码抄的。其实数据还行，虽然没跑赢沪深三百，但是确实赚钱了，贴一下 源地址, 大佬说有一部分巧合在里面。不多说了。

我结合下只买五块钱的情况

#!/usr/bin/python3    
# encoding: utf-8     
# @Time    : 2020/11/22 16:33    
# @author  : zza    
# @Email   : 740713651@qq.com    
# @File    : 闷声发小财。py    
import numpy    
import pandas    
from rqalpha.apis import *    
    
__config__ = {    
    "base": {    
        "start_date": "20190101",    
        "end_date": "20191231",    
        'frequency': '1d',    
        "accounts": {    
            "stock": 50000,    
        },    
        "data_bundle_path": r"E:\data\bundle",    
    },    
    "extra": {    
        "log_level": "debug",    
    },    
    "mod": {    
        "sys_progress": {    
            "enabled": True,    
            "show": True    
        }, "sys_accounts": {    
            "enabled": True,    
            "dividend_reinvestment": True,    
        },    
        "sys_analyser": {    
            "enabled": True,    
            "plot": True,    
            'benchmark': '000300.XSHG',    
        },    
    },    
}    
    
import talib    
    
def init(context):    
    context.s1 = '000300.XSHG'    
    
    context.SHORTPERIOD = 12    
    context.LONGPERIOD = 26    
    context.SMOOTHPERIOD = 9    
    context.OBSERVATION = 100    
    
    subscribe(context.s1)    
    context.lowest = 5    
    context.highest = 6    
    
    context.order_ticker = []    
    
def _history_bar_for_all_ins(x):    
    bar = history_bars(x, 1, "1d", fields=['close'])    
    if not bar:    
        return numpy.nan    
    if bar[0]:    
        return bar[0][0]    
    return numpy.nan    
    
def get_ticker(context):    
    all_ins = all_instruments("Stock").order_book_id.to_list()    
    df = pandas.DataFrame(all_ins, columns=['order_book_id'])    
    df['close'] = df.order_book_id.apply(_history_bar_for_all_ins)    
    price_df = df[(context.lowest <= df.close) & (df.close <= context.highest)]    
    
    order_ticker = []    
    for order_book_id in price_df.order_book_id.to_list():    
        c = pandas.Series(i[0] for i in history_bars(order_book_id, 3, "1d", fields=['close']))    
        if (c.shift(fill_value=0) < c).all():    
            order_ticker.append(order_book_id)    
    return order_ticker    
    
def before_trading(context):    
    print("年化：{}".format(context.portfolio.total_returns))    
    
def handle_bar(context, bar_dict):    
    closes = history_bars(context.s1, context.OBSERVATION, '1d', 'close')    
    
    diff, signal, _ = talib.MACD(closes)    
    
    if diff[-1] > signal[-1] and diff[-2] < signal[-2] :    
        if context.stock_account.cash > 1000:    
            context.order_ticker = get_ticker(context)    
            for item in context.order_ticker:    
                o = order_target_value(item, 2000)    
                if o:    
                    print(o)    
            print("context.order_ticker", context.order_ticker)    
    
    if diff[-1] < signal[-1] and diff[-2] > signal[-2]:    
        for item in context.portfolio.positions.keys():    
            order_target_percent(item, 0)    
    
def after_trading(context):    
    context.order_ticker = []    
if __name__ == '__main__':    
    import rqalpha    
    
    rqalpha.run_func(init=init,    
                     before_trading=before_trading,    
                     handle_bar=handle_bar,    
                     after_trading=after_trading,    
                     config=__config__)

image.png

还不如直接买华夏国运沪深三百。

朋友想法 A​

试试朋友的想法

假如设置一个均值（一直在变），低于 0.8 倍均值（mean_coe），并且连续 fall_days 天，跌幅加起来达到 fall_x%，买入 高于买入价格，并且连续 rise_days 天，涨幅加起来达到 rise_x%，卖出

我开始简单的设置了几个参数，发现结果不是很理想，但是没亏，我就想可能是参数的问题，然后列出了个集合，想着再跑跑，代码如下：

#!/usr/bin/python3    
# encoding: utf-8     
# @Time    : 2020/11/22 16:33    
# @author  : zza    
# @Email   : 740713651@qq.com    
import itertools    
    
import numpy    
import pandas    
from rqalpha.apis import *    
    
__config__ = {    
    "base": {    
        "start_date": "20190101",    
        "end_date": "20191231",    
        'frequency': '1d',    
        "accounts": {    
            "stock": 50000,    
        },    
        "data_bundle_path": r"E:\data\bundle",    
    },    
    "extra": {    
        "log_level": "debug",    
    },    
    "mod": {    
        "sys_progress": {    
            "enabled": True,    
            "show": True    
        }, "sys_accounts": {    
            "enabled": True,    
            "dividend_reinvestment": True,    
        },    
        "sys_analyser": {    
            "enabled": True,    
            "plot": True,    
            'benchmark': '000300.XSHG',    
        },    
    },    
}    
    
def init(context):    
    context.mean_coe = 0.8    
    context.mean_days = 244    
    context.rise_days = context.config.extra.rise_days    
    context.fall_days = context.config.extra.fall_days    
    context.rise_x = context.config.extra.rise_x    
    context.fall_x = context.config.extra.fall_x    
    context.order_ticker = [    
        "002111.XSHE",  # 威海广泰    
        "002673.XSHE",  # 西部证券    
        "601375.XSHG",  # 中原证券    
    ]    
    
    df = all_instruments('CS')    
    df = df[df.listed_date.apply(lambda x: x.year in [2015, 2016, 2017])]    
    context.order_ticker = context.order_ticker + df.order_book_id.to_list()    
    
def before_trading(context):    
    print("年化：{}".format(context.portfolio.total_returns))    
    # 价格均值    
    context.mean_close = {}    
    for order_book_id in context.order_ticker:    
        bars = pandas.Series([i[0] for i in history_bars(order_book_id, context.mean_days, "1d", fields=['close'])])    
        context.mean_close[order_book_id] = bars.mean()    
    
def handle_bar(context, bar_dict):    
    # for buy    
    if context.stock_account.cash >= 5000:    
    
        for order_book_id in context.order_ticker:    
            bar_close = bar_dict[order_book_id].close    
            if bar_close < (context.mean_close[order_book_id] * context.mean_coe):    
                _str = ""    
                bars = pandas.Series(    
                    [i[0] for i in history_bars(order_book_id, context.fall_days, "1d", fields=['close'])])    
                _str += "\n" + f"[{order_book_id}] 价格 ({bar_close}) 低于一年平均价{context.mean_close[order_book_id]}"    
                if not (bars.shift(-1, fill_value=0) < bars).all():    
                    # 判断连跌    
                    continue    
                _str += "\n" + f"[{order_book_id}] 连跌 价格表现为{list(bars)}"    
                if not (bars[0] * context.fall_x > bar_close):    
                    # 判断跌幅    
                    # 非低于 0.8 倍均值（mean_coe）    
                    continue    
                _str += "\n" + f"[{order_book_id}] 跌幅超过{context.fall_x}, {bars[0]} - {bar_close}"    
                o = order_value(order_book_id, 5000)    
                if o:    
                    _str += "\n" + f"[{order_book_id}] 买入".format(order_book_id)    
                    print(_str)    
    # for sell    
    for order_book_id in context.order_ticker:    
        position = context.portfolio.positions[order_book_id]    
        bar_close = bar_dict[order_book_id].close    
        if position is None or position.quantity <= 0:    
            continue    
        if position.avg_price < bar_close:    
            # 低于买入价格 不做操作    
            continue    
        _str = ""    
        bars = pandas.Series([i[0] for i in history_bars(order_book_id, context.fall_days, "1d", fields=['close'])])    
        if not (bars.shift(1, fill_value=0) < bars).all():    
            # 并且连续 rise_days 天    
            continue    
        _str += "\n" + f"[{order_book_id}] 连涨 价格表现为{list(bars)}"    
        if not (bars[0] * context.rise_x < bar_close):    
            # 涨幅加起来达到 rise_x%    
            continue    
        _str += "\n" + f"[{order_book_id}] 涨幅超过{context.rise_x}, {bars[0]} - {bar_close}"    
        o = order_target_percent(order_book_id, 0)    
        if o:    
            _str += "\n" + f"[{order_book_id}] 出售".format(order_book_id)    
            print(_str)    
    
def after_trading(context):    
    pass    
    
import rqalpha    
    
rise_days = range(3, 10)    
fall_days = range(3, 10)    
rise_x = numpy.arange(1.06, 1.12, 0.01)    
fall_x = numpy.arange(0.9, 0.94, 0.01)    
config_map = itertools.product(rise_days, fall_days, rise_x, fall_x)    
    
df = pandas.DataFrame(config_map, columns=["rise_days", "fall_days", "rise_x", "fall_x", ])    
result_list = []    
for _, item in df.iterrows():    
    __config__['extra']['rise_days'] = int(item.rise_days)    
    __config__['extra']['fall_days'] = int(item.fall_days)    
    __config__['extra']['rise_x'] = item.rise_x    
    __config__['extra']['fall_x'] = item.fall_x    
    result = rqalpha.run_func(init=init,    
                              before_trading=before_trading,    
                              handle_bar=handle_bar,    
                              after_trading=after_trading,    
                              config=__config__)    
    total_returns = result['sys_analyser']['summary']['total_value']    
    result_list.append(result)

其实跑出了几个超过大盘收益的结果。但是因为我的代码水平问题，报错了，结果也没保存，近期我会去吧这个跑完（1300 多的组合方式），有结果了我会贴上最优解的。

总结​

我还是不在股票市场里当韭菜了 溜了溜了

项目阅读需 10 分钟