Record

Annotated

from typing import Annotated
a = Annotated[int, "a"]
b = Annotated[int, "b"]

def func(a: a, b: b):
    pass

func(1, 2)
# TypeError: func() takes 0 positional arguments but 2 were given

Exception

try:
    # do something
except Exception as e:
    # do something
    raise e

# custom exception
class CustomException(Exception):
    # define custom exception to avoid catching other exceptions
    def __init__(self, message):
        self.message = message

# log exception
from loguru import logger

try:
    # do something
except Exception as e:
    logger.info(e)

fire

import fire
class Calculator(object):
    """A simple calculator class."""

    def double(self, number):
        return 2 * number

if __name__ == '__main__':
    fire.Fire(Calculator)

python calculator.py double 10

Generator

Compared with a list, the generator is more memory efficient.

import sys

range_size = range(1000)

def generator():
    for i in range_size:
        yield i

def list_generator():
    return [i for i in range_size]

print(sys.getsizeof(generator())) # 112
print(sys.getsizeof(list_generator())) # 9016

# List Comprehension
square_list = [i**2 for i in range(10)]
# Generator Expression
square_generator = (i**2 for i in range(10))

# Typical usegae for large data
def process_large_dataset(file_path):
    with open(file_path, 'r') as file:
        for line in file:
            # process_line() is a function that processes a single line2
            processed_data = process_line(line)
            yield processed_data

def process_line(line):
    # Conduct data processing logic, such as parsing, conversion, etc.
    # ...

# use generator to process large data
data_generator = process_large_dataset('large_data.txt')

# iterate over the generator
for processed_data in data_generator:
    # do something with processed_data
    print(processed_data)

inspect.getfullargspec

import inspect
def func(a, b, c=1, d=2, *args, **kwargs):
    pass
inspect.getfullargspec(func)
# FullArgSpec(args=['a', 'b', 'c', 'd'], varargs='args', varkw='kwargs', defaults=(1, 2), kwonlyargs=[], kwonlydefaults=None, annotations={})

logurn

from loguru import logger

logger.add(
    "{time:YYYY-MM-DD_HH:mm:ss}.log",
    level="INFO",
    rotation="1 day",
)

import package

For structure as below:

.
├── package1
│   ├── __init__.py
│   ├── module1.py
│   └── module2.py
└── package2
    ├── __init__.py
    ├── module3.py
    └── module4.py

If I want to import module1 in module3, I can use the following code:

import sys

sys.path.append("..")
from package1 import module1

Reference

os.path

import os
os.path.join("a", "b", "c")
# 'a/b/c'

os.path.normpath(os.path.join("a", "b", "c", ".."))
# 'a/b'

pandas

import pandas as pd

pd.set_option("display.max_columns", None)
pd.set_option("display.max_rows", None)
pd.set_option("display.width", 1000)

qlib.untils.auto_filter_kwargs

def auto_filter_kwargs(func: Callable, warning=True) -> Callable:
    def _func(*args, **kwargs):
        spec = inspect.getfullargspec(func)
        new_kwargs = {}
        for k, v in kwargs.items():
            # if `func` don't accept variable keyword arguments like `**kwargs` and have not according named arguments
            if spec.varkw is None and k not in spec.args:
                if warning:
                    log.warning(f"The parameter `{k}` with value `{v}` is ignored.")
            else:
                new_kwargs[k] = v
        return func(*args, **new_kwargs)

    return _func

@auto_filter_kwargs
def func(a, b, c=1, d=2, *args, **kwargs):
    pass

func(a=1, b=2, c=3, d=4, e=5, f=6)
# WARNING:root:The parameter `e` with value `5` is ignored.
# WARNING:root:The parameter `f` with value `6` is ignored.

rich.process

from rich.progress import Progress
from rich.progress import SpinnerColumn
from rich.progress import TimeElapsedColumn
from rich.progress import MofNCompleteColumn


progress = Progress(
    SpinnerColumn(),
    *Progress.get_default_columns(),
    TimeElapsedColumn(),
    MofNCompleteColumn(),
)


total = len(download_tasks)
market_data_task = progress.add_task(
    f"[cyan]Processing {batch_seqno}/{batch_num} batch", total=total
)

with progress:
    progress.update(market_data_task, completed=...)
    progress.update(market_data_task, advance=...)    

sqlalchemy

• Write str instead of list like ['a', 'b'] in filter method

data.trading_time_day = data.trading_time_day.apply(lambda x: f"'{x}'")
data.trading_time_night = data.trading_time_night.apply(lambda x: f"'{x}'")

shlex

import shlex

command = 'ls -l /path/to/directory "file with spaces.txt" -a'

# 使用 shlex.split() 分割命令
args = shlex.split(command)

print(args)

TypeDict

from typing import TypeDict
# TypeDict is a subclass of dict, which can be used as a type annotation.
class InstDict(TypeDict):
    var1: int
    var2: str
    var3: float

yaml

# load yaml
import yaml

# Specify the path to your YAML file
yaml_file_path = 'example.yaml'

# Open and read the YAML file
with open(yaml_file_path, 'r') as file:
    # Load YAML data
    data = yaml.safe_load(file)

# or
data = yaml.safe_load(open(yaml_file_path, 'r'))

# or
yaml_data = """
name: John Doe
age: 30
city: New York
"""

# Load YAML data
data = yaml.safe_load(yaml_data)

@dataclass

from dataclasses import dataclass
@dataclass
class A:
    a: int
    b: str
    c: float = 1.0
    d: int = 2
    e: str = "3"
    f: float = 4.0

A(1, "2")
# A(a=1, b='2', c=1.0, d=2, e='3', f=4.0)