-
Notifications
You must be signed in to change notification settings - Fork 0
/
model.py
190 lines (163 loc) · 6.08 KB
/
model.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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
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
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
150
151
152
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
182
183
184
185
186
187
188
189
190
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import numpy as np
def validateParams(sOn, sOff, herd, nAgents, echo=False):
"""
Function checks whether parameter set is valid. This is done by checking
the maximum probability, defiend as max(s) + h*N. If maximum probability
is equal to or greater than 1, then Error is raised. If maximum probability
is equal to or greater than 0.5, then Warning is raised.
Parameters
----------
sOn : float
sigma parameter for agents switching from "off" state to "on" state
sOff : float
sigma parameter for agents switching from "on" state to "off" state
herd : float
h parameter for interaction between agents
nAgents : int or float
number of agents within the model
Returns
-------
int
1 - everything is fine,
0 - warning
-1 - error
"""
maxSigma=np.max(np.array([sOn,sOff]))
maxProb=switchProb(maxSigma,herd,nAgents,nAgents)
if maxProb>=1:
if echo:
print(sOn,sOff,herd,nAgents,sep=",")
print("ERROR! Maximum probability larger than 1: {:.2f}".format(maxProb))
return -1
if maxProb>=0.5:
if echo:
print(sOn,sOff,herd,nAgents,sep=",")
print("WARNING! Maximum probability larger than 0.5: {:.2f}".format(maxProb))
return 0
if echo:
minSigma=np.min(np.array([sOn,sOff]))
minProb=switchProb(minSigma,herd,0,nAgents)
print(sOn,sOff,herd,nAgents,sep=",")
print("GOOD! Probability is within [{:.2f}; {:.2f}]".format(minProb,maxProb))
return 1
def switchProb(sigma,herd,nOthers,nTotal):
return sigma+herd*nOthers/nTotal
def step(sOn, sOff, herd, state):
"""
Function performs a single step according to herding model rules. Namely
switching probability to state ``i`` is assumed to be given by:
prob[i] = sigma[i] + h*N[i] ,
wher N[i] is the number of agents in ``i`` state.
Parameters
----------
sOn : float
sigma parameter for agents switching from "off" state to "on" state
sOff : float
sigma parameter for agents switching from "on" state to "off" state
herd : float
h parameter for interaction between agents
state : array_like
array containing states of all agents ("on" state is assumed to be
encoded by 1, while "off" state is assumed to be encoded by -1)
Returns
-------
array_like
updated state array
"""
# define internal state variable
_istate=state.copy()
# estimate current state properties
nTotal=len(_istate) # total number of agents
nOn=np.sum(_istate==1) # agents in on state
# calculate flipping probabilities
prob=np.zeros(nTotal)
prob[_istate==1]=switchProb(sOff,herd,nTotal-nOn,nTotal)
prob[_istate==-1]=switchProb(sOn,herd,nOn,nTotal)
# do the flip
r=np.random.rand(nTotal)
_istate[r<prob]=-_istate[r<prob]
return _istate
def rawSeries(nPoints, sOn, sOff, herd, state, warmup=0):
"""
Function performs multiple steps according to herding model rules (see
the documentation of step function).
Parameters
----------
nPoints : int
number of points to perform
sOn : float
sigma parameter for agents switching from "off" state to "on" state
sOff : float
sigma parameter for agents switching from "on" state to "off" state
herd : float
h parameter for interaction between agents
state : array_like
array containing states of all agents ("on" state is assumed to be
encoded by 1, while "off" state is assumed to be encoded by -1)
warmup : int, optional
number of warmup steps to perform (so that the model could have a
chance to "forget" the initial state). Zero by default.
Returns
-------
array_like
array of state observations
See also
--------
step
"""
# define internal state variable
_istate=state.copy()
# define output variable
output=np.zeros((nPoints, len(state)))
# the warmup loop
if warmup>0:
for i in np.arange(0, warmup):
_istate=step(sOn, sOff, herd, _istate)
# the main loop
for i in np.arange(0, nPoints):
_istate=step(sOn, sOff, herd, _istate)
output[i]=_istate.copy()
return output
def noisySeries(nPoints, pOn, pOff, sOn, sOff, herd, state, warmup=0):
"""
Function performs multiple steps according to herding model rules (see
the documentation of step function) and adds external noise. Namely, "on"
and "off" states now would mean that the states are noisy. Agent in "on"
state is on with probability given by pOn and agent in "off" state is on
with probability given by pOff.
Parameters
----------
nPoints : int
number of points to perform
pOn : float
probability that agent in the "on" state will be "on"
pOff : float
probability that agent in the "off" state will be "on"
sOn : float
sigma parameter for agents switching from "off" state to "on" state
sOff : float
sigma parameter for agents switching from "on" state to "off" state
herd : float
h parameter for interaction between agents
state : array_like
array containing states of all agents ("on" state is assumed to be
encoded by 1, while "off" state is assumed to be encoded by -1)
warmup : int, optional
number of warmup steps to perform (so that the model could have a
chance to "forget" the initial state). Zero by default.
Returns
-------
array_like
array of state observations (if parameters are valid) or array of None
(if parameters are not valid)
See also
--------
step, rawSeries
"""
series=rawSeries(nPoints, sOn, sOff, herd, state, warmup=warmup)
nseries=np.zeros(series.shape)
nseries[series<0]=(np.random.rand(np.sum(series<0))<pOff)
nseries[series>0]=(np.random.rand(np.sum(series>0))<pOn)
return nseries