Skip to content

signals

signals

Signal generation modules for player scoring.

FixtureSignal 

FixtureSignal(
    difficulty_ratings: Optional[dict[str, int]] = None,
)

Bases: BaseSignal

Generate signals based on fixture difficulty and schedule.

Initialize with team difficulty ratings.

PARAMETER DESCRIPTION
difficulty_ratings

Team strength ratings (1-5, higher = harder opponent)

TYPE: Optional[dict[str, int]] DEFAULT: None

Source code in fplx/signals/fixtures.py 
16
17
18
19
20
21
22
23
24
25
def __init__(self, difficulty_ratings: Optional[dict[str, int]] = None):
    """
    Initialize with team difficulty ratings.

    Parameters
    ----------
    difficulty_ratings : Optional[dict[str, int]]
        Team strength ratings (1-5, higher = harder opponent)
    """
    self.difficulty_ratings = difficulty_ratings or {}

generate_signal 

generate_signal(data)

Generate fixture-based signal.

Source code in fplx/signals/fixtures.py 
27
28
29
30
31
32
33
def generate_signal(self, data):
    """Generate fixture-based signal."""
    # This is a placeholder. The actual implementation would take
    # fixture data and compute a signal.
    return self.compute_fixture_advantage(
        data["team"], data["upcoming_opponents"], data["is_home"]
    )

set_difficulty_ratings 

set_difficulty_ratings(ratings: dict[str, int])

Set or update difficulty ratings.

PARAMETER DESCRIPTION
ratings

Team strength ratings

TYPE: Dict[str, int]

Source code in fplx/signals/fixtures.py 
35
36
37
38
39
40
41
42
43
44
def set_difficulty_ratings(self, ratings: dict[str, int]):
    """
    Set or update difficulty ratings.

    Parameters
    ----------
    ratings : Dict[str, int]
        Team strength ratings
    """
    self.difficulty_ratings = ratings

compute_fixture_difficulty 

compute_fixture_difficulty(
    team: str,
    upcoming_opponents: list[str],
    is_home: list[bool],
) -> float

Compute fixture difficulty score for upcoming games.

PARAMETER DESCRIPTION
team

Player's team

TYPE: str

upcoming_opponents

List of upcoming opponent teams

TYPE: list[str]

is_home

Whether each fixture is home

TYPE: list[bool]

RETURNS DESCRIPTION
float

Difficulty score (lower = easier fixtures)
Source code in fplx/signals/fixtures.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def compute_fixture_difficulty(
    self, team: str, upcoming_opponents: list[str], is_home: list[bool]
) -> float:
    """
    Compute fixture difficulty score for upcoming games.

    Parameters
    ----------
    team : str
        Player's team
    upcoming_opponents : list[str]
        List of upcoming opponent teams
    is_home : list[bool]
        Whether each fixture is home

    Returns
    -------
    float
        Difficulty score (lower = easier fixtures)
    """
    if not upcoming_opponents:
        return 3.0  # Neutral

    difficulties = []
    for opponent, home in zip(upcoming_opponents, is_home):
        # Get opponent difficulty
        diff = self.difficulty_ratings.get(opponent, 3)

        # Adjust for home advantage
        if home:
            diff = max(1, diff - 0.5)
        else:
            diff = min(5, diff + 0.5)

        difficulties.append(diff)

    # Average difficulty
    avg_difficulty = sum(difficulties) / len(difficulties)
    return avg_difficulty

compute_fixture_advantage 

compute_fixture_advantage(
    team: str,
    upcoming_opponents: list[str],
    is_home: list[bool],
) -> float

Compute fixture advantage (inverse of difficulty).

Higher score = easier fixtures = better for player.

PARAMETER DESCRIPTION
team

Player's team

TYPE: str

upcoming_opponents

List of upcoming opponent teams

TYPE: list[str]

is_home

Whether each fixture is home

TYPE: list[bool]

RETURNS DESCRIPTION
float

Advantage score (0-1, higher = better fixtures)
Source code in fplx/signals/fixtures.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def compute_fixture_advantage(
    self, team: str, upcoming_opponents: list[str], is_home: list[bool]
) -> float:
    """
    Compute fixture advantage (inverse of difficulty).

    Higher score = easier fixtures = better for player.

    Parameters
    ----------
    team : str
        Player's team
    upcoming_opponents : list[str]
        List of upcoming opponent teams
    is_home : list[bool]
        Whether each fixture is home

    Returns
    -------
    float
        Advantage score (0-1, higher = better fixtures)
    """
    difficulty = self.compute_fixture_difficulty(team, upcoming_opponents, is_home)

    # Convert to advantage (invert and normalize)
    # difficulty: 1 (easiest) to 5 (hardest)
    # advantage: 1 (best) to 0 (worst)
    advantage = (6 - difficulty) / 5
    return max(0, min(1, advantage))

compute_fixture_congestion 

compute_fixture_congestion(
    fixtures: DataFrame, team: str, days_window: int = 14
) -> float

Compute fixture congestion (number of games in short period).

PARAMETER DESCRIPTION
fixtures

Fixtures dataframe

TYPE: DataFrame

team

Team name

TYPE: str

days_window

Days to look ahead

TYPE: int DEFAULT: 14

RETURNS DESCRIPTION
float

Congestion score (0-1, higher = more congested)
Source code in fplx/signals/fixtures.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def compute_fixture_congestion(
    self, fixtures: pd.DataFrame, team: str, days_window: int = 14
) -> float:
    """
    Compute fixture congestion (number of games in short period).

    Parameters
    ----------
    fixtures : pd.DataFrame
        Fixtures dataframe
    team : str
        Team name
    days_window : int
        Days to look ahead

    Returns
    -------
    float
        Congestion score (0-1, higher = more congested)
    """
    # Filter fixtures for the team
    team_fixtures = fixtures[
        (fixtures["team_h"] == team) | (fixtures["team_a"] == team)
    ]

    if team_fixtures.empty:
        return 0.0

    # Count fixtures in window
    num_fixtures = len(team_fixtures)

    # Normalize: 1 game/week = 0, 3+ games/week = 1
    games_per_week = num_fixtures / (days_window / 7)
    congestion = min(1.0, (games_per_week - 1) / 2)

    return max(0, congestion)

batch_compute_advantages 

batch_compute_advantages(
    players_teams: dict[str, str],
    fixtures_data: dict[str, tuple],
) -> dict[str, float]

Compute fixture advantages for multiple players.

PARAMETER DESCRIPTION
players_teams

Mapping of player ID to team

TYPE: dict[str, str]

fixtures_data

Mapping of team to (opponents, is_home) tuples

TYPE: dict[str, tuple]

RETURNS DESCRIPTION
dict[str, float]

Dictionary of player fixture advantage scores
Source code in fplx/signals/fixtures.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def batch_compute_advantages(
    self, players_teams: dict[str, str], fixtures_data: dict[str, tuple]
) -> dict[str, float]:
    """
    Compute fixture advantages for multiple players.

    Parameters
    ----------
    players_teams : dict[str, str]
        Mapping of player ID to team
    fixtures_data : dict[str, tuple]
        Mapping of team to (opponents, is_home) tuples

    Returns
    -------
    dict[str, float]
        Dictionary of player fixture advantage scores
    """
    advantages = {}

    for player_id, team in players_teams.items():
        if team in fixtures_data:
            opponents, is_home = fixtures_data[team]
            advantage = self.compute_fixture_advantage(team, opponents, is_home)
            advantages[player_id] = advantage
        else:
            advantages[player_id] = 0.5  # Neutral

    return advantages

NewsParser

Parse and interpret FPL news text into structured signals.

parse_availability 

parse_availability(news_text: str) -> float

Parse availability from news text.

PARAMETER DESCRIPTION
news_text

News text

TYPE: str

RETURNS DESCRIPTION
float

Availability score (0-1)
Source code in fplx/signals/news.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def parse_availability(self, news_text: str) -> float:
    """
    Parse availability from news text.

    Parameters
    ----------
    news_text : str
        News text

    Returns
    -------
    float
        Availability score (0-1)
    """
    if not news_text or news_text.strip() == "":
        return 1.0

    text_lower = news_text.lower()

    # Check unavailable patterns
    for pattern in self.UNAVAILABLE_PATTERNS:
        if re.search(pattern, text_lower):
            return 0.0

    # Check doubtful patterns
    for pattern in self.DOUBTFUL_PATTERNS:
        if re.search(pattern, text_lower):
            return 0.5

    # Check positive patterns
    for pattern in self.POSITIVE_PATTERNS:
        if re.search(pattern, text_lower):
            return 0.9

    # Default: assume available if no negative signals
    return 1.0

parse_minutes_risk 

parse_minutes_risk(news_text: str) -> float

Parse minutes risk from news text.

PARAMETER DESCRIPTION
news_text

News text

TYPE: str

RETURNS DESCRIPTION
float

Minutes risk score (0-1, higher = more risk)
Source code in fplx/signals/news.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def parse_minutes_risk(self, news_text: str) -> float:
    """
    Parse minutes risk from news text.

    Parameters
    ----------
    news_text : str
        News text

    Returns
    -------
    float
        Minutes risk score (0-1, higher = more risk)
    """
    if not news_text or news_text.strip() == "":
        return 0.0

    text_lower = news_text.lower()

    # Check rotation patterns
    for pattern in self.ROTATION_PATTERNS:
        if re.search(pattern, text_lower):
            return 0.7

    # Check if doubtful (moderate risk)
    for pattern in self.DOUBTFUL_PATTERNS:
        if re.search(pattern, text_lower):
            return 0.3

    return 0.0

parse_confidence 

parse_confidence(news_text: str) -> float

Estimate confidence in the parsed signal.

PARAMETER DESCRIPTION
news_text

News text

TYPE: str

RETURNS DESCRIPTION
float

Confidence score (0-1)
Source code in fplx/signals/news.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def parse_confidence(self, news_text: str) -> float:
    """
    Estimate confidence in the parsed signal.

    Parameters
    ----------
    news_text : str
        News text

    Returns
    -------
    float
        Confidence score (0-1)
    """
    if not news_text or news_text.strip() == "":
        return 1.0  # High confidence when no news

    # Confidence based on clarity of news
    text_lower = news_text.lower()

    # High confidence patterns
    if any(
        re.search(p, text_lower) for p in ["ruled out", "confirmed", "definitely"]
    ):
        return 0.9

    # Medium confidence patterns
    if any(re.search(p, text_lower) for p in ["likely", "expected", "should"]):
        return 0.7

    # Low confidence patterns
    if any(re.search(p, text_lower) for p in ["maybe", "possible", "unclear"]):
        return 0.4

    return 0.6  # Default medium confidence

NewsSignal 

NewsSignal()

Bases: BaseSignal

Generate structured news signals for players.

Source code in fplx/signals/news.py 
155
156
def __init__(self):
    self.parser = NewsParser()

generate_signal 

generate_signal(news_text: str) -> dict[str, float]

Generate signal from news text.

PARAMETER DESCRIPTION
news_text

News text

TYPE: str

RETURNS DESCRIPTION
dict[str, float]

Dictionary with availability, minutes_risk, confidence
Source code in fplx/signals/news.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def generate_signal(self, news_text: str) -> dict[str, float]:
    """Generate signal from news text.

    Parameters
    ----------
    news_text : str
        News text

    Returns
    -------
    dict[str, float]
        Dictionary with availability, minutes_risk, confidence
    """
    availability = self.parser.parse_availability(news_text)
    minutes_risk = self.parser.parse_minutes_risk(news_text)
    confidence = self.parser.parse_confidence(news_text)

    return {
        "availability": availability,
        "minutes_risk": minutes_risk,
        "confidence": confidence,
        "adjustment_factor": availability * (1 - minutes_risk),
    }

batch_generate 

batch_generate(
    news_dict: dict[str, str],
) -> dict[str, dict[str, float]]

Generate signals for multiple players.

PARAMETER DESCRIPTION
news_dict

Dictionary mapping player ID to news text

TYPE: dict[str, str]

RETURNS DESCRIPTION
dict[str, dict[str, float]]

Dictionary of player signals
Source code in fplx/signals/news.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def batch_generate(self, news_dict: dict[str, str]) -> dict[str, dict[str, float]]:
    """
    Generate signals for multiple players.

    Parameters
    ----------
    news_dict : dict[str, str]
        Dictionary mapping player ID to news text

    Returns
    -------
    dict[str, dict[str, float]]
        Dictionary of player signals
    """
    signals = {}
    for player_id, news_text in news_dict.items():
        signals[player_id] = self.generate_signal(news_text)

    return signals

StatsSignal 

StatsSignal(weights: Optional[dict[str, float]] = None)

Generate performance signals from statistical data.

Combines multiple statistical indicators into a unified score.

Initialize with custom weights for different stats.

PARAMETER DESCRIPTION
weights

Weights for different statistics

TYPE: Optional[dict[str, float]] DEFAULT: None

Source code in fplx/signals/stats.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def __init__(self, weights: Optional[dict[str, float]] = None):
    """
    Initialize with custom weights for different stats.

    Parameters
    ----------
    weights : Optional[dict[str, float]]
        Weights for different statistics
    """
    self.weights = weights or {
        "points_mean": 0.3,
        "xG_mean": 0.15,
        "xA_mean": 0.15,
        "minutes_consistency": 0.2,
        "form_trend": 0.2,
    }

compute_signal 

compute_signal(player_data: DataFrame) -> float

Compute aggregated signal score from player statistics.

PARAMETER DESCRIPTION
player_data

Player historical data with engineered features

TYPE: DataFrame

RETURNS DESCRIPTION
float

Aggregated signal score (0-100)
Source code in fplx/signals/stats.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def compute_signal(self, player_data: pd.DataFrame) -> float:
    """
    Compute aggregated signal score from player statistics.

    Parameters
    ----------
    player_data : pd.DataFrame
        Player historical data with engineered features

    Returns
    -------
    float
        Aggregated signal score (0-100)
    """
    if player_data.empty:
        return 0.0

    # Get latest row (most recent data)
    latest = player_data.iloc[-1]

    score = 0.0

    # Points form (rolling mean)
    if "points_rolling_5_mean" in latest:
        points_component = (
            latest["points_rolling_5_mean"] * self.weights["points_mean"]
        )
        score += points_component

    # xG contribution
    if "xG_rolling_5_mean" in latest:
        xg_component = latest["xG_rolling_5_mean"] * 10 * self.weights["xG_mean"]
        score += xg_component

    # xA contribution
    if "xA_rolling_5_mean" in latest:
        xa_component = latest["xA_rolling_5_mean"] * 10 * self.weights["xA_mean"]
        score += xa_component

    # Minutes consistency (inverse of coefficient of variation)
    if "minutes_consistency_5" in latest:
        consistency = 1.0 / (1.0 + latest["minutes_consistency_5"])
        consistency_component = (
            consistency * 10 * self.weights["minutes_consistency"]
        )
        score += consistency_component

    # Form trend
    if "points_trend_5" in latest:
        trend = latest["points_trend_5"]
        # Normalize trend: positive trend is good
        trend_component = max(0, trend) * 5 * self.weights["form_trend"]
        score += trend_component

    return max(0, score)

batch_compute 

batch_compute(
    players_data: dict[str, DataFrame],
) -> dict[str, float]

Compute signals for multiple players.

PARAMETER DESCRIPTION
players_data

Dictionary mapping player ID/name to their data

TYPE: dict[str, DataFrame]

RETURNS DESCRIPTION
dict[str, float]

Dictionary of player signals
Source code in fplx/signals/stats.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def batch_compute(self, players_data: dict[str, pd.DataFrame]) -> dict[str, float]:
    """
    Compute signals for multiple players.

    Parameters
    ----------
    players_data : dict[str, pd.DataFrame]
        Dictionary mapping player ID/name to their data

    Returns
    -------
    dict[str, float]
        Dictionary of player signals
    """
    signals = {}
    for player_id, data in players_data.items():
        signals[player_id] = self.compute_signal(data)

    return signals