函数
函数是一段可以重复调用的代码。
例如,下面的代码重复执行了三次:
name = "张三"
print(f"你好,{name}")
name = "李四"
print(f"你好,{name}")
name = "王五"
print(f"你好,{name}")
可以将重复逻辑封装成函数:
def greet(name):
print(f"你好,{name}")
调用函数:
greet("张三")
greet("李四")
greet("王五")
函数的主要作用:
- 减少重复代码
- 提高代码可读性
- 方便维护和修改
- 将复杂问题拆分成小问题
- 便于测试和复用
使用 def 定义函数。
基本格式:
def 函数名(参数):
函数体
例如:
def say_hello():
print("Hello, Python")
调用:
say_hello()
运行结果:
Hello, Python
注意:
def是定义函数的关键字- 函数名后面必须有圆括号
- 行尾必须有冒号
- 函数体必须缩进
- 函数只有被调用时才会执行
函数名遵循变量命名规则。
推荐使用:
get_user_name()
calculate_total_price()
load_stock_data()
不推荐:
GetUserName()
calculateTotalPrice()
a()
Python 函数名通常使用:
小写字母 + 下划线
例如:
def get_stock_price():
pass
函数名应该尽量表达函数的作用。
推荐:
def calculate_profit():
pass
不推荐:
def process():
pass
因为 process 含义过于模糊。
有时暂时只想定义函数,但还没有编写具体逻辑,可以使用 pass。
def load_data():
pass
pass 表示什么都不做。
如果函数体为空,会出现语法错误:
def load_data():
因此可以先写:
def load_data():
pass
参数用于向函数传递数据。
def greet(name):
print(f"你好,{name}")
调用:
greet("张三")
其中:
name是形式参数,简称形参"张三"是实际参数,简称实参
def show_stock(code):
print(f"股票代码:{code}")
调用:
show_stock("600519")
结果:
股票代码:600519
def show_stock(code, name):
print(f"{code} {name}")
调用:
show_stock("600519", "贵州茅台")
参数会按照位置依次传递。
code = "600519"
name = "贵州茅台"
按照参数顺序传递的参数叫位置参数。
def introduce(name, age, city):
print(f"我叫{name},今年{age}岁,来自{city}")
调用:
introduce("张三", 25, "成都")
参数顺序错误,会导致结果错误:
introduce(25, "成都", "张三")
虽然代码可能不报错,但含义已经错误。
调用函数时,可以明确指定参数名:
introduce(
name="张三",
age=25,
city="成都",
)
使用关键字参数时,顺序可以改变:
introduce(
city="成都",
name="张三",
age=25,
)
结果不受顺序影响。
关键字参数适合:
- 参数较多
- 参数含义容易混淆
- 希望调用代码更清晰
例如:
def create_order(
code,
price,
quantity,
direction,
):
print(code, price, quantity, direction)
调用:
create_order(
code="600519",
price=1500.0,
quantity=100,
direction="buy",
)
参数可以设置默认值。
def greet(name, message="你好"):
print(f"{message},{name}")
调用时不传 message:
greet("张三")
结果:
你好,张三
调用时传入 message:
greet("张三", "早上好")
结果:
早上好,张三
正确:
def greet(name, message="你好"):
pass
错误:
def greet(message="你好", name):
pass
会出现语法错误。
正确规则:
必填参数在前
默认参数在后
例如文件读取:
def read_file(
file_path,
encoding="utf-8",
):
pass
调用:
read_file("data.txt")
或者:
read_file(
"data.txt",
encoding="gbk",
)
例如股票筛选:
def filter_stocks(
stocks,
min_price=0,
max_price=10000,
):
pass
函数可以通过 return 返回结果。
def add(a, b):
return a + b
调用:
result = add(10, 20)
print(result)
结果:
30
使用 print():
def add(a, b):
print(a + b)
调用:
result = add(10, 20)
print(result)
输出:
30
None
因为函数只是打印结果,没有返回结果。
使用 return:
def add(a, b):
return a + b
此时:
result = add(10, 20)
result 得到 30。
区别:
print → 只负责显示
return → 把结果交给函数外部
实际项目中,应优先返回结果,由调用者决定是否打印。
推荐:
def calculate_total(price, quantity):
return price * quantity
total = calculate_total(20, 5)
print(total)
def test():
print("第一行")
return
print("第二行")
调用:
test()
结果:
第一行
return 后面的代码不会执行。
提前返回可以减少代码嵌套。
不推荐:
def process_code(code):
if code:
if len(code) == 6:
if code.isdigit():
return code
return None
推荐:
def process_code(code):
if not code:
return None
if len(code) != 6:
return None
if not code.isdigit():
return None
return code
这种写法称为提前返回。
优点:
- 层级更少
- 逻辑更清晰
- 更容易维护
函数可以返回多个值:
def get_stock():
return "600519", "贵州茅台", 1500.0
调用:
result = get_stock()
print(result)
结果:
("600519", "贵州茅台", 1500.0)
实际上返回的是元组。
可以直接解包:
code, name, price = get_stock()
print(code)
print(name)
print(price)
如果函数没有写 return,默认返回 None。
def say_hello():
print("Hello")
result = say_hello()
print(result)
输出:
Hello
None
下面两个函数效果相同:
def test():
pass
def test():
return None
Python 3.12 支持类型提示。
def add(a: int, b: int) -> int:
return a + b
含义:
a: int → 参数 a 预期是整数
b: int → 参数 b 预期是整数
-> int → 返回值预期是整数
字符串参数:
def format_code(code: str) -> str:
return code.strip()
列表参数:
def total(numbers: list[int]) -> int:
return sum(numbers)
字典参数:
def show_stock(
stock: dict[str, object],
) -> None:
print(stock)
注意:
类型提示不会在运行时自动限制参数类型。
例如:
def add(a: int, b: int) -> int:
return a + b
下面仍然可能执行:
print(add("Hello", "Python"))
结果:
HelloPython
类型提示主要用于:
- 提高代码可读性
- 帮助编辑器检查错误
- 配合类型检查工具
- 明确函数接口
def normalize_name(name: str) -> str:
return name.strip()
def calculate_total(
price: float,
quantity: int,
) -> float:
return price * quantity
def is_valid_code(code: str) -> bool:
return len(code) == 6 and code.isdigit()
def show_message(message: str) -> None:
print(message)
None 表示函数不返回有意义的结果。
def filter_codes(
codes: list[str],
) -> list[str]:
return [
code
for code in codes
if code.startswith(("00", "60"))
]
def get_stock() -> dict[str, str]:
return {
"code": "600519",
"name": "贵州茅台",
}
def get_stock_info() -> tuple[str, str, float]:
return "600519", "贵州茅台", 1500.0
def get_unique_codes(
codes: list[str],
) -> set[str]:
return set(codes)
函数可能返回字符串,也可能返回 None:
def find_name(
user_id: int,
) -> str | None:
if user_id == 1:
return "张三"
return None
str | None 表示:
返回字符串,或者返回 None
def format_value(
value: int | float | str,
) -> str:
return str(value)
文档字符串用于说明函数用途。
def calculate_total(
price: float,
quantity: int,
) -> float:
"""
计算商品总价。
参数:
price: 商品单价。
quantity: 商品数量。
返回:
商品总价。
"""
return price * quantity
查看函数帮助:
help(calculate_total)
查看文档字符串:
print(calculate_total.__doc__)
简短函数可以使用单行文档字符串:
def add(a: int, b: int) -> int:
"""返回两个整数之和。"""
return a + b
如果函数需要接收任意数量的位置参数,可以使用 *args。
def total(*numbers):
return sum(numbers)
调用:
print(total(1, 2))
print(total(1, 2, 3))
print(total(1, 2, 3, 4, 5))
结果:
3
6
15
函数内部的 numbers 是元组。
def show_values(*values):
print(values)
print(type(values))
调用:
show_values(10, 20, 30)
结果:
(10, 20, 30)
<class 'tuple'>
args 只是约定名称,也可以写:
def total(*values):
return sum(values)
但通常推荐使用 args。
def show_scores(name, *scores):
print(f"姓名:{name}")
print(f"成绩:{scores}")
调用:
show_scores("张三", 80, 90, 85)
结果:
姓名:张三
成绩:(80, 90, 85)
如果函数需要接收任意数量的关键字参数,可以使用 **kwargs。
def show_user(**user):
print(user)
调用:
show_user(
name="张三",
age=25,
city="成都",
)
结果:
{
"name": "张三",
"age": 25,
"city": "成都",
}
函数内部的 user 是字典。
def show_values(**kwargs):
print(type(kwargs))
结果:
<class 'dict'>
def create_stock(code, **details):
print(f"代码:{code}")
print(details)
调用:
create_stock(
"600519",
name="贵州茅台",
price=1500.0,
industry="白酒",
)
Python 函数参数常见顺序:
def function(
普通参数,
默认参数,
*args,
仅限关键字参数,
**kwargs,
):
pass
例如:
def example(
a,
b=10,
*args,
c,
d=20,
**kwargs,
):
print(a)
print(b)
print(args)
print(c)
print(d)
print(kwargs)
调用:
example(
1,
2,
3,
4,
c=5,
d=6,
name="Python",
)
结果:
1
2
(3, 4)
5
6
{'name': 'Python'}
Python 可以使用 / 限制某些参数只能按位置传递。
def divide(a, b, /):
return a / b
正确:
divide(10, 2)
错误:
divide(a=10, b=2)
/ 前面的参数只能使用位置参数。
完整示例:
def calculate(a, b, /, operation="add"):
if operation == "add":
return a + b
return a - b
使用 * 可以限制后面的参数必须使用关键字传递。
def create_order(
code,
*,
price,
quantity,
):
print(code, price, quantity)
正确:
create_order(
"600519",
price=1500.0,
quantity=100,
)
错误:
create_order(
"600519",
1500.0,
100,
)
仅限关键字参数适合参数含义容易混淆的情况。
例如:
def connect_database(
host: str,
*,
port: int = 5432,
timeout: int = 10,
) -> None:
pass
调用更清晰:
connect_database(
"localhost",
port=5432,
timeout=30,
)
函数:
def add(a, b, c):
return a + b + c
列表:
numbers = [10, 20, 30]
普通调用:
result = add(
numbers[0],
numbers[1],
numbers[2],
)
使用解包:
result = add(*numbers)
相当于:
add(10, 20, 30)
元组也可以:
numbers = (10, 20, 30)
result = add(*numbers)
函数:
def show_stock(code, name, price):
print(code, name, price)
字典:
stock = {
"code": "600519",
"name": "贵州茅台",
"price": 1500.0,
}
解包调用:
show_stock(**stock)
相当于:
show_stock(
code="600519",
name="贵州茅台",
price=1500.0,
)
字典的键必须和函数参数名一致。
变量可以在不同范围中生效,这个范围称为作用域。
常见作用域:
- 局部作用域
- 全局作用域
- 嵌套作用域
- 内置作用域
Python 查找变量时遵循 LEGB 规则:
L:Local,局部作用域
E:Enclosing,外层函数作用域
G:Global,全局作用域
B:Built-in,内置作用域
函数内部定义的变量叫局部变量。
def test():
message = "Hello"
print(message)
message 只能在函数内部使用。
错误:
test()
print(message)
会出现:
NameError
函数外部定义的变量叫全局变量。
message = "Hello"
def show_message():
print(message)
函数内部可以读取全局变量:
show_message()
直接赋值时,Python 会把变量当作局部变量:
count = 0
def increase():
count = count + 1
会出现错误。
可以使用 global:
count = 0
def increase():
global count
count += 1
调用:
increase()
print(count)
结果:
1
但一般不推荐频繁使用 global。
更推荐传入参数并返回结果:
def increase(count: int) -> int:
return count + 1
count = 0
count = increase(count)
def outer():
message = "Hello"
def inner():
print(message)
inner()
内部函数可以读取外层函数的变量。
如果内部函数需要修改外层函数变量,可以使用 nonlocal。
def counter():
count = 0
def increase():
nonlocal count
count += 1
return count
return increase
使用:
increase = counter()
print(increase())
print(increase())
print(increase())
结果:
1
2
3
列表、字典、集合是可变对象。
函数中修改可变对象,会影响函数外部。
def add_item(items):
items.append("Python")
调用:
languages = ["Java"]
add_item(languages)
print(languages)
结果:
["Java", "Python"]
因为函数接收到的是同一个列表对象的引用。
如果不希望修改原列表,可以复制:
def add_item(items):
new_items = items.copy()
new_items.append("Python")
return new_items
调用:
languages = ["Java"]
new_languages = add_item(languages)
print(languages)
print(new_languages)
不要使用可变对象作为默认参数。
错误写法:
def add_item(
item,
items=[],
):
items.append(item)
return items
调用:
print(add_item("A"))
print(add_item("B"))
print(add_item("C"))
结果:
["A"]
["A", "B"]
["A", "B", "C"]
因为默认列表只创建一次,多次调用共享同一个列表。
正确写法:
def add_item(
item: str,
items: list[str] | None = None,
) -> list[str]:
if items is None:
items = []
items.append(item)
return items
调用:
print(add_item("A"))
print(add_item("B"))
print(add_item("C"))
结果:
["A"]
["B"]
["C"]
纯函数具有两个特点:
- 相同输入总是得到相同输出
- 不修改函数外部的数据
纯函数示例:
def add(a: int, b: int) -> int:
return a + b
非纯函数:
total = 0
def add_to_total(value: int) -> None:
global total
total += value
再例如:
def append_item(
items: list[str],
item: str,
) -> list[str]:
new_items = items.copy()
new_items.append(item)
return new_items
这个函数不修改原列表,更容易测试和维护。
在数据处理项目中,优先编写纯函数通常更安全。
Lambda 是一种简短的匿名函数。
普通函数:
def square(number):
return number ** 2
Lambda:
square = lambda number: number ** 2
调用:
print(square(5))
结果:
25
stocks = [
{
"name": "贵州茅台",
"price": 1500,
},
{
"name": "平安银行",
"price": 11,
},
{
"name": "宁德时代",
"price": 300,
},
]
按价格排序:
stocks.sort(
key=lambda stock: stock["price"],
)
按价格降序:
stocks.sort(
key=lambda stock: stock["price"],
reverse=True,
)
Lambda 只能包含一个表达式。
适合:
lambda x: x * 2
不适合复杂逻辑。
复杂逻辑应该定义普通函数:
def calculate_score(stock):
price = stock["price"]
growth = stock["growth"]
return growth / price
然后:
stocks.sort(key=calculate_score)
可读性更好。
Python 中,函数也是对象,可以作为参数传递。
def add(a, b):
return a + b
def subtract(a, b):
return a - b
定义接收函数的函数:
def calculate(a, b, operation):
return operation(a, b)
调用:
print(calculate(10, 5, add))
print(calculate(10, 5, subtract))
结果:
15
5
注意传递函数时不要加括号:
calculate(10, 5, add)
不是:
calculate(10, 5, add())
函数可以返回另一个函数。
def get_operation(operation):
if operation == "add":
return lambda a, b: a + b
if operation == "subtract":
return lambda a, b: a - b
raise ValueError("不支持的运算")
调用:
operation = get_operation("add")
print(operation(10, 20))
结果:
30
闭包是内部函数记住外部函数变量的一种机制。
def create_multiplier(factor):
def multiply(number):
return number * factor
return multiply
创建一个乘以 2 的函数:
double = create_multiplier(2)
创建一个乘以 10 的函数:
times_ten = create_multiplier(10)
调用:
print(double(5))
print(times_ten(5))
结果:
10
50
内部函数 multiply() 记住了外层函数的 factor。
闭包常用于:
- 保存状态
- 创建配置函数
- 装饰器
- 回调函数
装饰器用于在不修改原函数代码的情况下,增加额外功能。
例如:
def log_call(func):
def wrapper():
print("函数开始执行")
func()
print("函数执行结束")
return wrapper
原函数:
@log_call
def say_hello():
print("Hello")
调用:
say_hello()
结果:
函数开始执行
Hello
函数执行结束
下面两种写法效果相同:
@log_call
def say_hello():
print("Hello")
等价于:
def say_hello():
print("Hello")
say_hello = log_call(say_hello)
from functools import wraps
def log_call(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(f"调用函数:{func.__name__}")
result = func(*args, **kwargs)
print("函数执行完成")
return result
return wrapper
使用:
@log_call
def add(a: int, b: int) -> int:
return a + b
调用:
result = add(10, 20)
print(result)
结果:
调用函数:add
函数执行完成
30
@wraps(func) 用于保留原函数名称和文档信息。
函数调用自身,称为递归。
例如计算阶乘:
5! = 5 × 4 × 3 × 2 × 1
递归写法:
def factorial(number: int) -> int:
if number <= 1:
return 1
return number * factorial(number - 1)
调用:
print(factorial(5))
结果:
120
递归必须有终止条件:
if number <= 1:
return 1
否则会无限调用,最终出现:
RecursionError
普通循环通常更简单:
def factorial(number: int) -> int:
result = 1
for value in range(2, number + 1):
result *= value
return result
Python 不适合特别深的递归。
函数中使用 yield,会变成生成器函数。
def generate_numbers():
yield 1
yield 2
yield 3
调用:
generator = generate_numbers()
print(generator)
返回的是生成器对象。
遍历:
for number in generate_numbers():
print(number)
结果:
1
2
3
return:
- 返回结果
- 结束函数
yield:
- 暂停函数
- 返回一个值
- 下次继续执行
例如:
def count_up_to(limit: int):
number = 1
while number <= limit:
yield number
number += 1
使用:
for number in count_up_to(5):
print(number)
生成器适合处理大量数据,因为不会一次性把所有结果放入内存。
定义函数:
def add(a: int, b: int) -> int:
return a + b
查看类型注解:
print(add.__annotations__)
结果:
{
"a": int,
"b": int,
"return": int,
}
查看函数名称:
print(add.__name__)
查看文档字符串:
print(add.__doc__)
Python 3.12 可以使用 type 定义类型别名。
type StockCode = str
使用:
def normalize_code(
code: StockCode,
) -> StockCode:
return code.strip()
复杂类型别名:
type Stock = dict[str, object]
type StockList = list[Stock]
使用:
def filter_stocks(
stocks: StockList,
) -> StockList:
return stocks
Python 3.12 支持新的泛型函数语法。
def first[T](items: list[T]) -> T | None:
if not items:
return None
return items[0]
调用字符串列表:
names = ["张三", "李四"]
result = first(names)
返回类型会被理解为:
str | None
调用整数列表:
numbers = [10, 20, 30]
result = first(numbers)
返回类型会被理解为:
int | None
再例如:
def last[T](items: list[T]) -> T | None:
if not items:
return None
return items[-1]
不推荐:
def process_data():
# 读取文件
# 清洗数据
# 筛选股票
# 保存文件
# 发送消息
pass
推荐拆分:
def load_data():
pass
def clean_data(data):
pass
def filter_data(data):
pass
def save_data(data):
pass
主函数负责组合:
def main():
data = load_data()
cleaned_data = clean_data(data)
filtered_data = filter_data(cleaned_data)
save_data(filtered_data)
推荐:
load_data()
save_data()
calculate_profit()
filter_stocks()
validate_code()
不推荐:
data()
stock()
profit()
因为函数代表动作。
参数过多:
def create_user(
name,
age,
city,
phone,
email,
address,
company,
job,
):
pass
可以考虑使用字典:
def create_user(user: dict[str, object]):
pass
或者数据类:
from dataclasses import dataclass
@dataclass
class User:
name: str
age: int
city: str
phone: str
email: str
函数:
def create_user(user: User) -> None:
pass
不推荐:
tax_rate = 0.13
def calculate_tax(price):
return price * tax_rate
更推荐:
def calculate_tax(
price: float,
tax_rate: float,
) -> float:
return price * tax_rate
调用时更明确:
tax = calculate_tax(
price=100,
tax_rate=0.13,
)
不推荐:
def find_stock(code):
if code == "600519":
return {
"code": "600519",
"name": "贵州茅台",
}
return False
返回字典和布尔值混合,不利于理解。
推荐:
def find_stock(
code: str,
) -> dict[str, str] | None:
if code == "600519":
return {
"code": "600519",
"name": "贵州茅台",
}
return None
不推荐:
def calculate_total():
price = float(input("请输入价格:"))
quantity = int(input("请输入数量:"))
return price * quantity
推荐:
def calculate_total(
price: float,
quantity: int,
) -> float:
return price * quantity
输入放在外部:
price = float(input("请输入价格:"))
quantity = int(input("请输入数量:"))
total = calculate_total(price, quantity)
print(total)
这样函数更容易测试。
def divide(a: float, b: float) -> float:
if b == 0:
raise ValueError("除数不能为零")
return a / b
调用:
try:
result = divide(10, 0)
except ValueError as error:
print(error)
def validate_stock_code(code: str) -> str:
if not isinstance(code, str):
raise TypeError("code 必须是字符串")
code = code.strip()
if len(code) != 6:
raise ValueError("股票代码长度必须为 6 位")
if not code.isdigit():
raise ValueError("股票代码必须全部由数字组成")
return code
调用:
try:
code = validate_stock_code(" 600519 ")
except (TypeError, ValueError) as error:
print(error)
else:
print(code)
不推荐:
def divide(a, b):
try:
return a / b
except Exception:
return None
这样会隐藏所有问题。
更推荐:
def divide(a: float, b: float) -> float:
if b == 0:
raise ValueError("除数不能为零")
return a / b
由调用者决定如何处理异常。
需求:
- 参数必须是字符串
- 长度必须为 6 位
- 必须全部由数字组成
- 以
00开头,返回33-00 - 以
60开头,返回17-60 - 其他代码返回原代码
def convert_stock_code(code: str) -> str:
"""
根据股票代码前缀转换代码。
规则:
00 开头转换为 33-00。
60 开头转换为 17-60。
其他代码保持不变。
"""
if not isinstance(code, str):
raise TypeError("code 必须是字符串")
code = code.strip()
if len(code) != 6:
raise ValueError("code 长度必须为 6 位")
if not code.isdigit():
raise ValueError("code 必须全部由数字组成")
if code.startswith("00"):
return "33-00"
if code.startswith("60"):
return "17-60"
return code
调用:
codes = [
"000001",
"600519",
"300750",
]
for code in codes:
result = convert_stock_code(code)
print(code, result)
结果:
000001 33-00
600519 17-60
300750 300750
原始数据:
stocks = [
{
"code": "600519",
"name": "贵州茅台",
"price": 1500.0,
},
{
"code": "000001",
"name": "平安银行",
"price": 11.5,
},
{
"code": "300750",
"name": "宁德时代",
"price": 300.0,
},
]
定义筛选函数:
def filter_stocks(
stocks: list[dict[str, object]],
*,
prefixes: tuple[str, ...] = ("00", "60"),
min_price: float = 0,
max_price: float | None = None,
) -> list[dict[str, object]]:
"""
根据代码前缀和价格范围筛选股票。
"""
result: list[dict[str, object]] = []
for stock in stocks:
code = str(stock.get("code", ""))
price = float(stock.get("price", 0))
if not code.startswith(prefixes):
continue
if price < min_price:
continue
if max_price is not None and price > max_price:
continue
result.append(stock)
return result
调用:
result = filter_stocks(
stocks,
prefixes=("00", "60"),
min_price=10,
max_price=1000,
)
print(result)
结果只包含平安银行。
def count_code_prefixes(
codes: list[str],
length: int = 2,
) -> dict[str, int]:
"""
统计股票代码前缀出现次数。
"""
if length <= 0:
raise ValueError("length 必须大于 0")
counts: dict[str, int] = {}
for code in codes:
clean_code = code.strip()
if len(clean_code) < length:
continue
prefix = clean_code[:length]
counts[prefix] = counts.get(prefix, 0) + 1
return counts
调用:
codes = [
"600519",
"603259",
"000001",
"002594",
"300750",
"688981",
]
result = count_code_prefixes(codes)
print(result)
结果:
{
"60": 2,
"00": 2,
"30": 1,
"68": 1,
}
将复杂任务拆分为多个函数。
from pathlib import Path
import json
读取 JSON:
def load_json(
file_path: Path,
) -> list[dict[str, object]]:
text = file_path.read_text(
encoding="utf-8",
)
data = json.loads(text)
if not isinstance(data, list):
raise ValueError("JSON 根节点必须是列表")
return data
清理数据:
def clean_stocks(
stocks: list[dict[str, object]],
) -> list[dict[str, object]]:
result: list[dict[str, object]] = []
for stock in stocks:
clean_stock = stock.copy()
clean_stock["code"] = str(
stock.get("code", "")
).strip()
clean_stock["name"] = str(
stock.get("name", "")
).strip()
result.append(clean_stock)
return result
按照代码去重:
def deduplicate_stocks(
stocks: list[dict[str, object]],
) -> list[dict[str, object]]:
result: list[dict[str, object]] = []
seen_codes: set[str] = set()
for stock in stocks:
code = str(stock.get("code", ""))
if code in seen_codes:
continue
seen_codes.add(code)
result.append(stock)
return result
筛选主板股票:
def filter_main_board(
stocks: list[dict[str, object]],
) -> list[dict[str, object]]:
return [
stock
for stock in stocks
if str(
stock.get("code", "")
).startswith(("00", "60"))
]
保存 JSON:
def save_json(
file_path: Path,
data: list[dict[str, object]],
) -> None:
text = json.dumps(
data,
ensure_ascii=False,
indent=2,
)
file_path.write_text(
text,
encoding="utf-8",
)
主函数:
def main() -> None:
input_path = Path("stocks.json")
output_path = Path("main_board_stocks.json")
stocks = load_json(input_path)
stocks = clean_stocks(stocks)
stocks = deduplicate_stocks(stocks)
stocks = filter_main_board(stocks)
save_json(output_path, stocks)
print(f"共保存 {len(stocks)} 条数据")
程序入口:
if __name__ == "__main__":
main()
这个程序的处理流程:
读取数据
→ 清理数据
→ 去重
→ 筛选
→ 保存
每个函数只负责一个任务。
函数应该尽量方便测试。
例如:
def add(a: int, b: int) -> int:
return a + b
简单测试:
assert add(1, 2) == 3
assert add(-1, 1) == 0
assert add(0, 0) == 0
如果条件不成立,会出现:
AssertionError
测试股票代码:
def is_valid_stock_code(code: str) -> bool:
return (
len(code) == 6
and code.isdigit()
)
测试:
assert is_valid_stock_code("600519") is True
assert is_valid_stock_code("000001") is True
assert is_valid_stock_code("60051") is False
assert is_valid_stock_code("ABC123") is False
def say_hello():
print("Hello")
这只是定义函数,并不会输出。
需要调用:
say_hello()
函数:
def add(a, b):
return a + b
错误调用:
add(1)
或者:
add(1, 2, 3)
def calculate_total(price, quantity):
total = price * quantity
调用:
result = calculate_total(10, 5)
print(result)
结果:
None
正确:
def calculate_total(price, quantity):
total = price * quantity
return total
错误理解:
def add(a, b):
print(a + b)
函数不能把打印结果继续交给其他代码使用。
推荐:
def add(a, b):
return a + b
不推荐:
result = []
def add_item(item):
result.append(item)
推荐:
def add_item(
items: list[str],
item: str,
) -> list[str]:
return [*items, item]
不推荐:
def add_item(item, items=[]):
items.append(item)
return items
推荐:
def add_item(
item,
items=None,
):
if items is None:
items = []
items.append(item)
return items
如果函数达到几十行甚至上百行,通常需要考虑拆分。
例如:
def process_stocks():
# 读取
# 校验
# 清理
# 去重
# 统计
# 排序
# 保存
pass
可以拆分成多个小函数。
不推荐:
def find_stock(code):
if code == "600519":
return {
"code": code,
}
return False
推荐:
def find_stock(
code: str,
) -> dict[str, str] | None:
if code == "600519":
return {
"code": code,
}
return None
编写一个函数,接收两个数字,返回较大的数字。
参考答案:
def get_larger(
a: float,
b: float,
) -> float:
if a >= b:
return a
return b
也可以:
def get_larger(
a: float,
b: float,
) -> float:
return max(a, b)
编写函数判断一个数字是否为偶数。
def is_even(number: int) -> bool:
return number % 2 == 0
编写函数计算列表平均值。
def calculate_average(
numbers: list[float],
) -> float:
if not numbers:
raise ValueError("列表不能为空")
return sum(numbers) / len(numbers)
编写函数,筛选列表中的偶数。
def filter_even_numbers(
numbers: list[int],
) -> list[int]:
return [
number
for number in numbers
if number % 2 == 0
]
编写函数,判断股票代码是否合法。
规则:
- 必须是字符串
- 长度为 6
- 全部是数字
def is_valid_stock_code(code: str) -> bool:
return (
isinstance(code, str)
and len(code) == 6
and code.isdigit()
)
编写函数,统计字符串中每个字符出现次数。
def count_characters(
text: str,
) -> dict[str, int]:
counts: dict[str, int] = {}
for char in text:
counts[char] = counts.get(char, 0) + 1
return counts
编写函数,根据股票价格排序。
def sort_stocks_by_price(
stocks: list[dict[str, object]],
*,
reverse: bool = False,
) -> list[dict[str, object]]:
return sorted(
stocks,
key=lambda stock: float(
stock.get("price", 0)
),
reverse=reverse,
)
编写函数,返回列表中的第一个和最后一个元素。
def get_first_and_last[T](
items: list[T],
) -> tuple[T, T]:
if not items:
raise ValueError("列表不能为空")
return items[0], items[-1]
定义函数:
def function_name():
pass
带参数:
def greet(name):
print(name)
带返回值:
def add(a, b):
return a + b
默认参数:
def greet(name, message="你好"):
pass
可变位置参数:
def total(*numbers):
return sum(numbers)
可变关键字参数:
def show_user(**user):
print(user)
类型提示:
def add(a: int, b: int) -> int:
return a + b
可选返回值:
def find_name(user_id: int) -> str | None:
pass
仅限关键字参数:
def create_order(
code,
*,
price,
quantity,
):
pass
列表解包:
function(*items)
字典解包:
function(**data)
函数设计时应重点考虑:
函数负责什么?
需要哪些输入?
返回什么结果?
是否修改外部数据?
是否需要参数验证?
是否容易单独测试?
一个清晰函数通常具有以下特征:
函数名明确
参数数量适中
只完成一个任务
返回值稳定
尽量不依赖全局变量
包含类型提示
必要时包含文档字符串
函数的本质是:
输入数据
→ 执行处理
→ 返回结果
掌握函数后,就可以把较长的 Python 程序拆分成多个清晰、独立、可复用的模块。