18h_daily_job.py 文件源码

def stat_today_all(tmp_datetime):
    datetime_str = (tmp_datetime).strftime("%Y-%m-%d")
    datetime_int = (tmp_datetime).strftime("%Y%m%d")
    print("datetime_str:", datetime_str)
    print("datetime_int:", datetime_int)
    data = ts.get_today_all()
    # ?????????????????????????concat????
    if not data is None and len(data) > 0:
        # ??????
        # del data["reason"]
        data["date"] = datetime_int  # ??????int???
        data = data.drop_duplicates(subset="code", keep="last")
        data.head(n=1)
        common.insert_db(data, "ts_today_all", False, "`date`,`code`")
    else:
        print("no data .")

    time.sleep(5)  # ??5?

    data = ts.get_index()
    # ?????????????????????????concat????
    if not data is None and len(data) > 0:
        # ??????
        # del data["reason"]
        data["date"] = datetime_int  # ??????int???
        data = data.drop_duplicates(subset="code", keep="last")
        data.head(n=1)
        common.insert_db(data, "ts_index_all", False, "`date`,`code`")
    else:
        print("no data .")

    print(datetime_str)


# main????