魔のkyo的BLOG

# -*- coding: utf-8 -*-
import sys
import os
from functools import cmp_to_key


def mycmp_filename(name1, name2):
    if name1.upper() == "README.MD":
        return -1
    elif name2.upper() == "README.MD":
        return 1
    elif name1 < name2:
        return -1
    elif name2 < name1:
        return 1
    return 0


def gen(e, path, level):
    content = ""
    name, ext = os.path.splitext(e)
    if os.path.isfile(path):
        if not name.startswith('_') and ext.lower() == ".md":
            if os.path.getsize(path):
                content += "{}- [{}]({})\n".format("  "*level, name, path.replace('\\', '/') )
    elif os.path.isdir(path):
        if name.startswith('_'):
           return "" 
        content2 = ""
        for e2 in sorted(os.listdir(path), key=cmp_to_key(mycmp_filename)):
            content2 += gen(e2, os.path.join(path, e2), level+1)
        if content2:
            if name:
                content += "{}- {}\n".format("  "*level, name)
            content += content2
        else:
            content = ""
    return content


def main():
    with open("_sidebar.md", "w", encoding="utf-8") as fp:
        fp.write("<!-- docs/_sidebar.md -->\n")
        fp.write("{}\n".format(gen("", ".", -1)))

if __name__ == '__main__':
    main()

import re
import json

# 删除非ascii的编码
def omit_unascii(data):
    new_data = b''
    p = 0
    m = re.search(b"[^\x00-\x7F]+", data[p:])
    while m:
        new_data += data[p:p+m.start()]
        p += m.end()
        m = re.search(b"[^\x00-\x7F]+", data[p:])
    new_data += data[p:]
    return new_data


def json_loads(msg):
    try:
        # 先尝试正常解析（按UTF8解码）
        obj = json.loads(msg)
    except Exception as e:
        try:
            # 尝试先按GBK解码再解析
            obj = json.loads(msg.decode("gb2312", "ignore"))
        except Exception as e:
            try:
                # 尝试删除所有非ascii编码后再解析
                obj = json.loads(omit_unascii(msg))
            except Exception as e:
                raise e
    return obj

# -*- coding: utf-8 -*-
import pymongo
import threading


class _HashedSeq(list):
    __slots__ = 'hashvalue'

    def __init__(self, tup, hash=hash):
        self[:] = tup
        self.hashvalue = hash(tup)

    def __hash__(self):
        return self.hashvalue


def _make_key(args, kw, typed=False):
    key = args
    if kw:
        kwd_mark = (object(),)
        key += kwd_mark
        for item in kw.items():
            key += item
    if typed:
        key += tuple(type(v) for v in args)
        if kw:
            key += tuple(type(v) for v in kw.values())
    elif len(key) == 1 and type(key[0]) in {int, str}:
        return key[0]
    return _HashedSeq(key)


local = threading.local()

class MongoDB:
    @staticmethod
    def instance(*args, **kw):
        if not hasattr(local, "mongodb") or local.mongodb is None:
            local.mongodb = {}
        key = _make_key(args, kw, typed=False)
        if key not in local.mongodb:
            local.mongodb[key] = MongoDB(*args, **kw)
        return local.mongodb[key]

    def __init__(self, host="127.0.0.1", port=27017, dbname=None, username=None, password=None, **kw):
        self.client = pymongo.MongoClient(f"mongodb://{host}:{port}/{dbname}", username=username, password=password, **kw)
        if dbname:
            self.db = self.client[dbname]
        else:
            self.db = self.client.test

    def close(self):
        return self.client.close()

    def __getattr__(self, key):
        return self.db[key]

    def __getitem__(self, key):
        return self.db[key]

    def has_collection(self, name):
        return name in self.db.list_collection_names()


# 常用运算符
# 逻辑运算
# {$or:[expression1,expression2,...]}
# {$and:[expression1,expression2,...]}
# {$not:expression1}}
# 比较运算
# {field:{$eq:value}}  ==
# {field:{$ne:value}}  !=
# {field:{$lt:value}}  <
# {field:{$lte:value}} <=
# {field:{$gt:value}}  >
# {field:{$gte:value}} >=
# {field:{$in:[value1,value2,...]} in
# {field:{$nin:[value1,value2,...]} not in
# 正则匹配
# {field:{"$regex": "正则表达式"}

# query: 查询条件, ex: {"type": "stock"}
# fields: 提取字段, ex: ["code", "sec_name"]
# sort: 排序字段和排序方式, ex: [("code", -1)] 表示按code排倒序
# limit: 之取前limit个, ex: 10
# **kw: 查询条件，把query展开成参数来写，作用是相同的
def get_securities(query={}, fields=None, sort=None, limit=None, **kw):
    q = gen_query("securities", query, fields, sort, **kw)
    return [doc for doc in (q.limit(limit) if limit else q)]

def gen_query(collection, query={}, fields=None, sort=None, **kw):
    mongo = MongoDB.instance(host="192.168.1.99", dbname="ftresearch", username="ftresearch", password="******")
    q = mongo[collection].find(dict(**query, **kw), to_projection_dict(fields))
    if sort:
        q = q.sort(to_sort_list(sort))
    return q


def to_projection_dict(fields):
    if fields is None:
        return {"_id": 0}
    elif isinstance(fields, dict):
        return fields
    else:
        ret = {"_id": 0}
        for k in fields:
            ret[k] = 1
        return ret


def to_sort_list(sort):
    if not sort:
        return None
    ret = []
    if isinstance(sort, list):
        for e in sort:
            if isinstance(e, str):
                ret.append((e, 1))
            else:
                ret.append(e)
    else:
        if isinstance(sort, str):
            ret.append((sort, 1))
        else:
            ret.append(sort)
    return ret


if __name__ == '__main__':
    # 查询以68编码开头的所有股票，只取code, sec_name字段
    arr = get_securities(code={"$regex": "^68"}, type="stock", fields=["code", "sec_name"])
    for e in arr:
        print(e)

    # 查询所有ST状态和*ST状态的未退市的股票信息，按股票代码排倒序，取前10只
    arr = get_securities(
        query={
        "type": "stock",
        "contract_state": "Active",
        "special_type": {"$in": ["ST", "StarST"]},
        },
        fields=["code", "sec_name", "exchange", "special_type"],
        sort=[("code", -1)],
        limit=10)
    for e in arr:
        print(e)

    # 取2023-01-04日当天在交易状态的股指期货合约代码
    codes = [e["code"]
        for e in get_securities(
            fields=["code"],
            type="future",
            product={"$in": ["IC", "IF", "IH", "IM"]},
            listed_date={"$lte":"2023-01-04"},
            de_listed_date={"$gte":"2023-01-04"}
            )
        ]
    print(codes)

from functools import cmp_to_key

score = [97, 80, 90, 63, 72, 84]
idx = range(len(score))

# cmp传入的是类似C语言qsort的比较函数，返回负、零、正来表示大小关系
idx = sorted(idx, key=cmp_to_key(lambda i, j: score[i]-score[j]))
print(idx) # idx的序是按照score[idx[i]]的值排的
sorted_score = [score[i] for i in idx]
assert(sorted_score==sorted(score))
print(sorted_score)

import os
import platform
import traceback


def exec(cmd:str):
    pip = os.popen(cmd)
    r = pip.buffer.read()
    try:
        return r.decode(encoding='utf8')
    except Exception as e:
        try:
            return r.decode(encoding='gb2312')
        except Exception as e:
            return r
    return r


def is_mounted(path):
    if platform.system() == 'Windows':
        return os.path.exists(path)
    else:
        return int(exec(f"df -h|grep {path}|wc -l")) > 0

import math

def coord_dist(lat1, lnt1, lat2, lnt2):
    sin = math.sin
    cos = math.cos
    acos = math.acos
    deg2rad = math.radians

    R = 63713930
    lat1 = deg2rad(lat1)
    lnt1 = deg2rad(lnt1)
    lat2 = deg2rad(lat2)
    lnt2 = deg2rad(lnt2)
    l = R * acos(cos(lnt1)*cos(lnt2)*cos(lat1-lat2)+sin(lnt1)*sin(lnt2))
    return l

from croniter import croniter
from datetime import datetime
from dateutil import tz

now = datetime.now().replace(tzinfo=tz.gettz())  # 得到当前时间（带时区）
cron = croniter('0/5 9-14 * * 1-5', now) # 初始化cron对象

dt = cron.get_next(datetime) # 取得下次触发时间，实际上做的是迭代器去next()然后返回
print(dt)
dt = cron.get_next(datetime) # 所在第二次调用的时候，迭代器继续next()，这里返回的是第二次触发的时间
print(dt)
dt = cron.get_prev(datetime) # 同理，这里将返回第一次触发的时间，而不是初始化时间之前的上一次触发时间
print(dt)

cron = croniter('0/5 9-14 * * 1-5 H(0-59)', now, hash_id="hash_id")