SQLAlchemy查询：筛选最新周期值大于过往周期最新值的记录

表结构 #

class Valuation(Base):
    __tablename__ = 'valuation'
    id = Column(Integer, primary_key=True)
    reference = Column(BigInteger, index=True)
    value = Column(Float)
    period = Column(String)

示例数据 #

数据说明 #

• 并非所有reference在每个不同的周期序列（年份-字符）中都有value记录，某个reference可能没有某周期最新序列的value值。
• value应随时间递减或保持不变，因此任意周期的最大值应小于之前周期的最大值。

需求 #

筛选出所有满足以下条件的reference：该reference的最新周期value值大于其任何过往周期的最新value值。

示例中应返回结果：

现有代码问题 #

尝试使用aliased方法，但现有代码无法实现“最新周期值与每个过往周期最新值对比”的逻辑，且适配性极差：

value2022 = aliased(Valuation, name="value2022")
value2021 = aliased(Valuation, name="value2021")
query = (
    db.query(Valuation)
    .outerjoin(value2022, (
            (Valuation.reference == value2022.reference)
            & (Valuation.value > value2022.value)
            & (Valuation.period.startswith("2023"))
            & (value2022.period.startswith("2022"))
        )
    )
    .outerjoin(value2021, (
            (Valuation.reference == value2021.reference)
            & (Valuation.value > value2021.value)
            & (Valuation.period.startswith("2023"))
            & (value2021.period.startswith("2021"))
        )
    )
    .order_by(
        Valuation.reference,
        Valuation.period.desc(),
    )
    .distinct(Valuation.reference)
    .all()
)

核心思路是先聚合每个reference每年的最新周期value，再对比其最新年份的value是否大于所有过往年份的最大value，全程动态处理年份，无需硬编码。

方案一：子查询分步实现 #

from sqlalchemy import func, desc, and_
from sqlalchemy.orm import aliased

# 1. 子查询：获取每个reference+年份的最新周期
yearly_latest_period = db.query(
    Valuation.reference,
    func.substr(Valuation.period, 1, 4).label('year'),
    func.max(Valuation.period).label('latest_period')
).group_by(Valuation.reference, func.substr(Valuation.period, 1, 4)).subquery()

# 2. 子查询：获取每个reference每年最新周期对应的value
yearly_latest_value = db.query(
    Valuation.reference,
    Valuation.value.label('yearly_max_value'),
    func.substr(Valuation.period, 1, 4).label('year')
).join(
    yearly_latest_period,
    and_(
        Valuation.reference == yearly_latest_period.c.reference,
        Valuation.period == yearly_latest_period.c.latest_period
    )
).subquery()

# 3. 子查询：获取每个reference的最新年份
latest_year_per_ref = db.query(
    yearly_latest_value.c.reference,
    func.max(yearly_latest_value.c.year).label('latest_year')
).group_by(yearly_latest_value.c.reference).subquery()

# 4. 子查询：获取每个reference最新年份的value和对应period
latest_value_per_ref = db.query(
    yearly_latest_value.c.reference,
    yearly_latest_value.c.yearly_max_value.label('latest_value'),
    Valuation.period.label('latest_period')
).join(
    latest_year_per_ref,
    and_(
        yearly_latest_value.c.reference == latest_year_per_ref.c.reference,
        yearly_latest_value.c.year == latest_year_per_ref.c.latest_year
    )
).join(
    Valuation,
    and_(
        Valuation.reference == yearly_latest_value.c.reference,
        Valuation.value == yearly_latest_value.c.yearly_max_value,
        func.substr(Valuation.period, 1, 4) == yearly_latest_value.c.year
    )
).subquery()

# 5. 子查询：获取每个reference过往年份的最大value
max_past_yearly_value = db.query(
    yearly_latest_value.c.reference,
    func.max(yearly_latest_value.c.yearly_max_value).label('max_past_value')
).join(
    latest_year_per_ref,
    and_(
        yearly_latest_value.c.reference == latest_year_per_ref.c.reference,
        yearly_latest_value.c.year != latest_year_per_ref.c.latest_year
    )
).group_by(yearly_latest_value.c.reference).subquery()

# 最终查询：筛选符合条件的记录
result = db.query(
    latest_value_per_ref.c.reference,
    latest_value_per_ref.c.latest_value.label('value'),
    latest_value_per_ref.c.latest_period.label('period')
).outerjoin(
    max_past_yearly_value,
    latest_value_per_ref.c.reference == max_past_yearly_value.c.reference
).filter(
    # 兼容无过往记录的情况：默认过往最大值为负无穷
    latest_value_per_ref.c.latest_value > func.coalesce(max_past_yearly_value.c.max_past_value, -float('inf'))
).all()

方案二：窗口函数简化实现（需SQL支持窗口函数） #

如果使用PostgreSQL、MySQL 8+等支持窗口函数的数据库，可以用更简洁的代码实现：

from sqlalchemy import over, partition_by, func

# 1. 窗口函数：标记每个reference+年份内的最新周期记录
ranked_periods = db.query(
    Valuation,
    func.rank().over(
        partition_by=[Valuation.reference, func.substr(Valuation.period,1,4)],
        order_by=Valuation.period.desc()
    ).label('rank_in_year')
).subquery()

# 2. 提取每个reference每年的最新value
yearly_latest = db.query(
    ranked_periods.c.reference,
    ranked_periods.c.value.label('yearly_value'),
    func.substr(ranked_periods.c.period,1,4).label('year'),
    ranked_periods.c.period
).filter(ranked_periods.c.rank_in_year == 1).subquery()

# 3. 窗口函数：标记最新年份记录，并计算过往年份的最大value
ranked_years = db.query(
    yearly_latest,
    func.rank().over(
        partition_by=yearly_latest.c.reference,
        order_by=yearly_latest.c.year.desc()
    ).label('rank_over_years'),
    func.max(yearly_latest.c.yearly_value).over(
        partition_by=yearly_latest.c.reference,
        order_by=yearly_latest.c.year.desc(),
        rows=func.between(1, func.unbounded_following)
    ).label('max_past_value')
).subquery()

# 4. 筛选最终结果
result = db.query(
    ranked_years.c.reference,
    ranked_years.c.yearly_value.label('value'),
    ranked_years.c.period.label('period')
).filter(
    ranked_years.c.rank_over_years == 1,
    ranked_years.c.yearly_value > func.coalesce(ranked_years.c.max_past_value, -float('inf'))
).all()

逻辑说明 #

• 通过substr(period,1,4)动态提取年份，兼容任意年份的周期格式
• 先聚合每年的最新周期value，避免同一年份多条记录干扰对比
• 用coalesce处理无过往记录的reference，确保这类记录被正确筛选
• 两种方案均无需硬编码年份，适配性强，可应对新增年份的场景

内容的提问来源于stack exchange，提问作者Turukawa